How To Create Own Database?
test
В сучасному програмууванні дуже багато абстракцій. Ми користуємось фреймворками ,ORM системами. В цілому хороший програміст має знати якій набор інструментаріїв треба взяти під проєкт. Велика частина роботи це правильно інтегрувати допоміжні системи одна з одною. Схоже на лего, правда?
Так само і бд, це дуже складні проєкти, які виконують купу роботи, і все це ховається за звичайним select * from user
.
За основною задачею , бази даних мають виконувати дві речі. Зберігати передану до неї інформацію і видавати її по запиту. Найпростіше порівняння - бібліотека.. Є купа поличок (файли) і бібліотекар який знає де шукати.
Сучасні бд звичайно суттєво більш складні. При великому бажанні, можливо реалізувати великий production ready бізнес проєкт виключно засобами що нам дають бази даних. Але тема нашої розмови не про портування DOOM на холодильник
Мені стало цікаво як в базовому вигляді може працювати база даних, це і буде темою цих рядків.
Як часто кажуть, для того щоб зрозуміти як працюють ORM , треба зробити власну. За цією логікою, ми маємо створити власну бд
Примітка: Код написан виключно для демонстрації роботи і він не претендує на високу ефкутивність, чи хочаб на те щоб бути гарним.
Реалізуємо дуже просту БД з таким функціоналом:
- Створити таблицю
- Вставити запис до таблиці
- Взяти запис
- Зробимо індексацію по id
- База має бути енергонезалежною
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"}
Все працює. Таким чином ми створили власну енергонезалежну базу даних . Яка хоч і з натяжкою, але може виконувати свої задачі. Перші бази даних були не суттєво складнішими.
Як бачимо, основна таблиця записує записи в тому порядку як їх до неї відправляють. В той самий час таблиця індексів кожний раз сортує свої записи. Для того щоб в майбутньмо при читанні з цієї таблиці можна було користуватись алгоритмом бінарного пошуку. Само собою це дуже спрощена поведінка.
Rejection Of The Dependency Manager Pipenv
When the dependency manager pipenv became popular and its included in PyPA GitHub organization. We had been starting to use pipenv in our projects both in development and in a deploymen
What makes startups fail
All founders need to have a certain level of belief to start a company, a confidence that they have hit on the sort of idea that could grow, that will become profitable, and ultimately succeed via an IPO. But the harsh reality of the startup landscape paints a different picture, with nearly 70% of all tech startups failing.
Digital banking as a mainstream in FinTech - are you ready?
Any bank that is not ready for the digitalization should prepare to keep up with rivals. The competition is strong and demand is high. There are a lot of already transformed financial institutions into successful mobile banks and they raise their influence every day, as they operate faster and provide smarter and more efficient services.