Get Started

How To Create Own Database?

test

В сучасному програмууванні дуже багато абстракцій. Ми користуємось фреймворками ,ORM системами. В цілому хороший програміст має знати якій набор інструментаріїв треба взяти під проєкт. Велика частина роботи це правильно інтегрувати допоміжні системи одна з одною. Схоже на лего, правда?

Так само і бд, це дуже складні проєкти, які виконують купу роботи, і все це ховається за звичайним select * from user.

За основною задачею , бази даних мають виконувати дві речі. Зберігати передану до неї інформацію і видавати її по запиту. Найпростіше порівняння - бібліотека.. Є купа поличок (файли) і бібліотекар який знає де шукати.

Сучасні бд звичайно суттєво більш складні. При великому бажанні, можливо реалізувати великий production ready бізнес проєкт виключно засобами що нам дають бази даних. Але тема нашої розмови не про портування DOOM на холодильник

Мені стало цікаво як в базовому вигляді може працювати база даних, це і буде темою цих рядків.

Як часто кажуть, для того щоб зрозуміти як працюють ORM , треба зробити власну. За цією логікою, ми маємо створити власну бд

Примітка: Код написан виключно для демонстрації роботи і він не претендує на високу ефкутивність, чи хочаб на те щоб бути гарним.

Реалізуємо дуже просту БД з таким функціоналом:

  1. Створити таблицю
  2. Вставити запис до таблиці
  3. Взяти запис
  4. Зробимо індексацію по id
  5. База має бути енергонезалежною

Project Structure:

├── app
│   ├── db
│   │   ├──tables
│   │   └──indexes
│   ├── file_system
│   │   ├──crud.py
│   │   └──index .py
│   ├── lang
│   │   ├──_base.py
│   │   ├──_commander.py
│   │   ├──create_table.py
│   │   ├──select.py
│   │   └──inser.py
│   └── main.py
└──

db - файли даних, про якусь оптімізацію тут не говоримо, одна таблиця - один файл.

file_system - методи для роботи з файловою системою

lang - використовується для роботи з sql запитами

Далі буде код з описом Нажаль, розбити цей код на маленьки шматки з детальним описом занадто роздуло би розмір статті. Сподіваюсь що код достатньо говорячий.

main:

from lang._commander import LangCommander


def server():
    while True:
        if raw_command := input("naviSql : "):
            LangCommander.run_command(raw_command)


def main():
    server()


if __name__ == '__main__':
    main()

Програма у нескінченому циклі очікує команду від користувача. Все що тут робиться - передача до LangCommander всього що ввів користувач

lang._commander:

from lang.create_table import CreateTable
from lang.insert import Insert
from lang.select import Select


class LangCommander:

    @classmethod
    def run_command(cls, stringed_data: str) -> any:
        commands = {
            "insert": Insert,
            "select": Select,
            "create_table": CreateTable,
        }
        try:
            command = stringed_data[:stringed_data.index(" ")]

            ins = commands[command](stringed_data)
            print(ins.exec() or "no result")
        except Exception as exc:
            print(exc)

run_command мапить назву команди з сласом який відповідая за обробку Також Це наш вивід даних) Не знав що з ними робити, то нехай тут вже будуть 😃

class String:
    @classmethod
    def between(cls, command: str, start: str, finish: str) -> str:
        between_str = ((command.split(start))[1].split(finish)[0]).strip()
        return between_str

    @classmethod
    def right(cls, command: str, word: str) -> str:
        values_i = command.rfind(word)
        text = command[values_i + len(word):].strip()
        return text

**lang._base:**

Просто дві функції що постійно використовуються для розбору запита на його частини.

lang._commander:

from typing import Optional

from file_system.crud import CRUD


class BaseCommand:
    fs = CRUD
    find = String

    @classmethod
    def deconstruct(cls, stringed_data: str, mapper: Optional[dict[str, callable]] = None) -> dict:
        query = {}
        if not mapper:
            return query

        for i in stringed_data.split(" "):
            try:
                query[i] = mapper[i](stringed_data)
            except KeyError:
                pass
        return query

    @classmethod
    def exec(cls):
        raise NotImplementedError

Це базовий метод deconstruct(), який отримує сиру строчку шукає в ній якусь команду і звіряє це з маппером що йому передають.

На вході

{
"select": sql_select (func)
"from": sql_from (func)
}

На виході

{
"select": ["column1", "column2"]
"from": "table_name"
}

lang.create_table:

from typing import Optional
from lang._base import BaseCommand


class CreateTable(BaseCommand):

    def __init__(self, stringed_data: str):
        mapped_q = self.deconstruct(stringed_data)

        self.table_name: str = mapped_q["create_table"]
        self.table_columns: list = mapped_q["columns"]

    def exec(self) -> any:
        self.fs.create_index(self.table_name)
        return self.fs.create_table(self.table_name, self.table_columns)

    @classmethod
    def sql_create_table(cls, command: str) -> str:
        table_name = cls.find.between(command, "create_table", "columns")
        return table_name

    @classmethod
    def sql_columns(cls, command: str) -> list:
        columns = cls.find.right(command, "columns")
        return columns.split(",")

    @classmethod
    def deconstruct(cls, stringed_data: str, mapper: Optional[dict[str, callable]] = None) -> dict:
        commands_mapper = {
            "create_table": cls.sql_create_table,
            "columns": cls.sql_columns,
        }
        return super().deconstruct(stringed_data, commands_mapper)

lang.insert:

from typing import Optional
from lang._base import BaseCommand


class Insert(BaseCommand):

    def __init__(self, stringed_data: str):
        mapped_q = self.deconstruct(stringed_data)

        self.table_name = mapped_q["insert"]
        self.obj_data = mapped_q["values"]

    def exec(self) -> any:
        return self.fs.write(self.table_name, self.obj_data)

    @classmethod
    def sql_insert(cls, command: str) -> str:
        table_name = cls.find.between(command, "insert", "values")
        return table_name

    @classmethod
    def sql_values(cls, command: str) -> list:
        values = cls.find.right(command, "values")
        return values.split(",")

    @classmethod
    def deconstruct(cls, stringed_data: str, mapper: Optional[dict[str, callable]] = None) -> dict:
        commands_mapper = {
            "insert": cls.sql_insert,
            "values": cls.sql_values,
        }
        return super().deconstruct(stringed_data, commands_mapper)

lang.select:

from typing import Optional
from lang._base import BaseCommand


class Select(BaseCommand):

    def __init__(self, stringed_data: str):
        mapped_q = self.deconstruct(stringed_data)

        self.columns: list = mapped_q["select"]
        self.table_name: str = mapped_q["from"]
        self.obj_id: int = mapped_q["where"]

    def exec(self) -> any:
        return self.fs.read(self.table_name, self.obj_id, self.columns)


    @classmethod
    def sql_select(cls, command: str) -> list:
        args = cls.find.between(command, "select", "from")
        return args.split(",")

    @classmethod
    def sql_from(cls, command: str) -> str:
        table_name = cls.find.between(command, "from", "where")
        return table_name

    @classmethod
    def sql_where(cls, command: str) -> int:
        return int(cls.find.right(command, "where"))

    @classmethod
    def deconstruct(cls, stringed_data: str, mapper: Optional[dict[str, callable]] = None) -> dict:
        commands_mapper = {
            "select": cls.sql_select,
            "from": cls.sql_from,
            "where": cls.sql_where,
        }
        return super().deconstruct(stringed_data, commands_mapper)

В create_table, insert та select ми розбиваємо строку на необхідні елементи і за допомогою єдиної команди exec виконуємо її.

Кожний клас тут має специфічні для себе методи деконструкції типу sql_select, sql_insert Тут в методі deconstruct ми зьираємо необхідні функції для обробки команди та відправляємо це все в батьківський deconstruct При опрацьованні сирої строчки по черзі виконується необхідна функція. Не будемо вдивлятись в те що ми під кожний метод відправляємо всю строчку))

Доречі всі lang. класи були повністю перероблені в процесі написання статті. Спочатку це пряцювало за приципом вкладеності, де кожен зовнішній клас брав свою частину і вних передавав залишок. Але це було не наочно, то переробив. Гарна новина, в нас вийшла не дуже погана архітектура, бо в процесі не прийшлось міняти нічого з інших модулів. )

file_system.crud:

import json
from typing import Optional

from file_system.index import Index


class CRUD:
    index = Index
    DB_SOURCE = "db/tables"

    @classmethod
    def get_full_table_name(cls, table_name: str) -> str:
        return f"{cls.DB_SOURCE}/{table_name}"

    @classmethod
    def create_index(cls, table_name: str) -> None:
        cls.index.create_index(table_name)

    @classmethod
    def create_table(cls, table_name: str, obj_columns: list) -> None:
        with open(cls.get_full_table_name(table_name), "w") as f:
            f.write(f'{"___".join(obj_columns)}\n')

    @classmethod
    def read(cls, table_name: str, obj_id: int, columns: list) -> Optional[str]:

        if obj_index := cls.index.get_index(table_name, obj_id):

            with open(cls.get_full_table_name(table_name), "r") as f:
                lines = f.readlines()
                labels = lines[0].replace("\n", "").split("___")

                res = {}
                obj = lines[obj_index].replace("\n", "").split("___")

                column_indexes = cls.index.get_column_indexes(columns, labels)
                for index in column_indexes:
                    res[labels[index]] = obj[index]

                return json.dumps(res)

    @classmethod
    def count(cls, table_name: str) -> int:
        with open(cls.index.get_full_index_name(table_name), "r") as f:
            return len(f.readlines())

    @classmethod
    def write(cls, table_name: str, data: list):
        with open(cls.get_full_table_name(table_name), "a") as f:
            f.write(f'{"___".join(data)}\n')

        cls.index.update_index(table_name, data[0], cls.count(table_name))

Якщо тут ігнорувати роботу з індексацією, то це вкрай простий підхід. Ми можемо читати і писати в файлову систему.

Якщо ігнорувати той факт що дійсним базам даних необхідно піклуватись ще про купу речей (права, конкурентність, стабільність, перевовненість файлів, різні принципи що зпрощують життя розробникам). То в нас вийщла досить єфективна реалізація. При записі ми беремо необхідну таблицю і пишемо в кінець файлу новий запис. це просто але єфективно, запис в кінець файлу не дорона операція.

Але все ламається в момент коли нам необхідно щось отримати. Уявимо що в нас 10 мільйонів записів (забудемо що в нас тільки один файл для кожної таблиці:) ). Для того щоб знайти необхідне, в гіршому випадку придеться пройтись абсолютно по всім записам тобто O(n)

Тому додано індексування, хочаб захардкожене на id.

file_system.index:

import re
from bisect import bisect_left


class Index:
    DB_SOURCE = "db/indexes"

    @classmethod
    def get_full_index_name(cls, table_name: str) -> str:
        return f"{cls.DB_SOURCE}/{table_name}"

    @classmethod
    def create_index(cls, table_name: str) -> None:
        with open(cls.get_full_index_name(table_name), "w") as f:
            pass

    @classmethod
    def get_column_indexes(cls, selected: list, full: list) -> list:
        res = []
        for i_f, val_f in enumerate(full):
            for val_s in selected:
                if val_f == val_s:
                    res.append(i_f)
        return res

    @classmethod
    def get_index(cls, table_name: str, obj_id: int) -> int:
        with open(cls.get_full_index_name(table_name), "r") as f:
            index = cls.binary_search(f.readlines(), obj_id) + 1
            return index

    @classmethod
    def update_index(cls, table_name: str, obj_id: int, last_el: int):

        with open(cls.get_full_index_name(table_name), "r+") as f:
            data = f.read().splitlines()
            data.append(f"{obj_id}___{last_el}")
            data.sort(key=lambda s: int(re.search(r'\d+', s).group()))
            f.seek(0)
            f.writelines([line + "\n" for line in data])

    @classmethod
    def binary_search(cls, obj_list: list, obj_id: int) -> int:
        i = bisect_left([obj.split("___")[0] for obj in obj_list], str(obj_id))
        if i != len(obj_list) and obj_list[i].startswith(str(obj_id)):
            result = re.search('___(.*)\n', obj_list[i])
            return int(result.group(1))
        else:
            return -1

Індекс - в спрощеному вигляді це додаткова структура що має сортований список записів і посилання на їх реальне розташування.

За його принципом. Створюється окрема таблиця. В який зберішаються відсортовані за ID пари obj_id___obj_position. Це дозволяє в подальшому використовувати алгоритми швидкого погуку.

Впровадження індексів дозволяє суттєво підняти швидкість читання , правда ціною швидкості запису бо кожний раз при додаванні нового єлементу ми переписуємо весь файл індексів (в нашій реалізації) В реальних бд, це робиться інакше. Є дуже багато реалізацій за принципом, В деякиї випадках індекси розтащовані в файловій системі, в деякиї в оперативній памʼяті чи комбінованих варіантах.

Тут же, при додаванні нового запису CRUD посилає команду на перерахування файлу індексу.

Давайте протестуємо

Cпочатку створимо таблицю

naviSql : create_table projects columns id, name
no result
naviSql :

В директорії 'db' зʼявилось два нових файла. Пустий індекс та сама таблиця, де першою строчкою знаїодяться імена колонок накшталт csv: id___name .

db
├──tables
│  └──projects
├──indexes
│  └──projects

Додамо декілька записів

naviSql : insert projects values 1, cool project
no result
naviSql : insert projects values 87, cool project 87
no result
naviSql : insert projects values 23, cool project 23
no result
naviSql : insert projects values 11, cool project 11
no result
naviSql :

Давайте перевіримо як виглядає бд

db.tables.projects

id___name
1___cool project
87___cool project 87
23___cool project 23
11___cool project 11

db.tables.indexes

1___0
11___3
23___2
87___1

Ну і отримаємо данні

naviSql : select id, name from projects where 1
{"id": "1", "name": "cool project"}
naviSql : select id, name from projects where 23
{"id": "23", "name": "cool project 23"}
naviSql : select id, name from projects where 87
{"id": "87", "name": "cool project 87"}
naviSql :
naviSql : select name from projects where 87
{"name": "cool project 87"}

naviSql : select id from projects where 87
{"id": "87"}

Все працює. Таким чином ми створили власну енергонезалежну базу даних . Яка хоч і з натяжкою, але може виконувати свої задачі. Перші бази даних були не суттєво складнішими.

Як бачимо, основна таблиця записує записи в тому порядку як їх до неї відправляють. В той самий час таблиця індексів кожний раз сортує свої записи. Для того щоб в майбутньмо при читанні з цієї таблиці можна було користуватись алгоритмом бінарного пошуку. Само собою це дуже спрощена поведінка.

14 DEC, 2017

Django vs. Aiohttp Performance Test

Today I would like to take a trendy Python framework Django and compare it with a brand new framework aiohttp. I'm a Django fan and believe that Django is a really good framework, and I recommend to use it.

16 OCT, 2023

6 keys for an effective onboarding in small distributive team

How to build effective onboarding in terms of remote culture, rapid team growing and changing without the ability to delegate to personal onboarding buddies?

25 APR, 2018

Creating SVG Animation With Lottie

Designers often face the fact that if they create a cool icon / illustration on the site, it’s important to animate it exactly as it was intended. Many are afraid of the process itself, how can it be implemented independently without the involvement of a

Help Ukraine to stop russian aggression