ORM, или объектно-реляционное отображение — это программная технология, которая позволяет взаимодействовать с базами данных с использованием объектно-ориентированной парадигмы. Вместо того чтобы писать SQL-запросы напрямую для работы с данными в базе данных, можно использовать ORM, чтобы взаимодействовать с данными, как если бы они были объектами в вашем коде.
Не бывало ли вам интересно, как работает изнутри такая идейно простая концепция? Благодаря чему достигается удобство работы? Сегодня мы напишем ORM самостоятельно и узнаем, какие инструменты python нам для этого понадобятся.
Ремарка: sqlite3 выбран из-за простоты, нетрудно заменить обращения к нему на обращения к любой удобной для вас базе данных. По синтаксису я ориентировался на джанго.
▍ Базовые типы данных
В sqlite3 существуют: INTEGER — вещественное число с указанной точностью, TEXT — текст, BLOB — двоичные данные, REAL — число с плавающей запятой(float24), NUMERIC — то же, что и INTEGER.
CREATE TABLE "example" ( "Field1" INTEGER NOT NULL, "Field2" TEXT UNIQUE, "Field3" BLOB, "Field4" REAL DEFAULT 123, "Field5" NUMERIC ); У каждого из них есть параметры NULL, UNIQUE, DEFAULT, так что первым делом пишем класс, который будут наследовать все остальные:
class BaseType: field_type: str #название типа данных поля, например, "INTEGER" def __init__(self, unique: bool = False, null: bool = True, default: int = None): self.unique = unique self.null = null self.default = defaultНа основе него прописываем остальные базовые классы:
class IntegerField(BaseType): field_type = 'INTEGER' class TextField(BaseType): field_type = 'TEXT' class BlobField(BaseType): field_type = 'BLOB' class RealField(BaseType): field_type = 'REAL' class NumericField(BaseType): field_type = 'NUMERIC'
▍ Пользовательские модели
Я хочу, чтобы пользовательские модели выглядел максимально просто, например, так:
class Box(models.Model): name = TextField() width = IntegerField() height = IntegerField()Для этого реализуем родительский класс Model. Он должен задавать объекты с заданными переменными. Для этого пишем инициализатор:
class Model: def __init__(self, *args, **kwargs): fields = [el for el in vars(self.__class__) if not el.startswith("__")] #поля, которые мы создали в модели (в данном случае name, width, height) for i, value in enumerate(args): setattr(self, fields[i], value)#задаем переменные переданные с помощью args for field, value in kwargs.items():#задаем переменные переданные с помощью kwargs setattr(self, field, value)Все методы, которые, мы напишем для класса Model будут работать с любым объектом, который мы зададим.
— *args позволяет передавать произвольное количество позиционных аргументов в функцию.
— Аргументы, переданные как *args, собираются в кортеж (tuple) внутри функции.
— Вы можете использовать любое имя вместо «args», но общепринято использовать именно «args».
Пример:
def print_args(*args): for arg in args: print(arg) print_args(1, 2, 3) # Выводит: 1, 2, 3 2. **kwargs:
— **kwargs позволяет передавать произвольное количество именованных аргументов (ключ-значение) в функцию.
— Аргументы, переданные как **kwargs, собираются в словарь (dictionary) внутри функции.
— Вы можете использовать любое имя вместо «kwargs», но общепринято использовать именно «kwargs».
Пример:
def print_kwargs(**kwargs): for key, value in kwargs.items(): print(f"{key}: {value}") print_kwargs(name="John", age=30, city="New York") # Выводит: name: John, age: 30, city: New YorkИспользуя *args и **kwargs, вы можете создавать более гибкие функции, которые могут обрабатывать разные наборы аргументов.
Давайте также создадим сразу метод json(), чтобы возвращать объект в виде словаря, он понадобится нам, например, для api или удобного вывода в консоль.
def json(self): attributes = {} for key, value in vars(self).items(): if not key.startswith("__") and not callable(value):#проверка на системные методы и поля attributes[key] = value return attributesВ данный момент мы уже можем пользоваться моделями, но пока что без базы данных:
class Box(models.Model): name = TextField() width = IntegerField() height = IntegerField() print(Box('Box 1', 1, 1).json())#выведет {'name': 'Box 1', 'width': 1, 'height': 1}
▍ Добавляем sqlite3
Для использования базы данных я хочу, чтобы каждый объект имел поле objects — менеджер объектов, через который мы и будем обращаться к sqlite.
Например, так:
Box.objects.add(Box('BOX 1', 1, 1)) Box.objects.get(name='BOX 1') Box.objects.filter(width=1, height=1) Box.objects.delete(name="BOX 1")Мы хотим, чтобы каждый объект имел это поле, но при этом поведение различалось в зависимости от модели объекта. Для решения этого создадим прокси-класс, который определяет модель и возвращает нужный менеджер.
class ProxyObjects: def __get__(self, instance, owner): return Object(owner)
__get__ определяется внутри класса, который также может иметь методы __set__ и __delete__, если необходимо управлять операциями присваивания и удаления атрибутов.
Пример использования __get__:
class MyDescriptor: def __get__(self, instance, owner): if instance is None: # Если доступ к дескриптору осуществляется через класс, а не через экземпляр, # то instance будет равен None, и мы можем вернуть сам дескриптор или другое значение. return self else: # В этом случае instance - это экземпляр объекта, owner - это класс, к которому относится атрибут. # Мы можем вернуть значение атрибута или выполнить другие действия при доступе к нему. return instance._value class MyClass: def __init__(self, value): self._value = value # Используем дескриптор MyDescriptor для атрибута 'my_attribute' my_attribute = MyDescriptor() # Создаём экземпляр класса obj = MyClass(42) # Доступ к атрибуту 'my_attribute' будет вызывать метод __get__ дескриптора MyDescriptor print(obj.my_attribute) # Выведет: 42В этом примере MyDescriptor является дескриптором, который определяет поведение при доступе к атрибуту my_attribute класса MyClass. Метод __get__ определяет, что происходит при чтении значения этого атрибута через экземпляр obj.
Первое, что нужно сделать для работы с базой данных — создать таблицу. Для этого пишем метод:
import sqlite3 # Необходимо импортировать библиотеку для работы с SQLite class Object: def __init__(self, object_type: type): # Конструктор класса принимает тип объекта (класс) и сохраняет его в атрибуте object_type. self.object_type = object_type def __createTable__(self): # Метод для создания таблицы в базе данных, основанной на атрибутах класса object_type. # Устанавливаем соединение с базой данных conn = sqlite3.connect(db_name) cursor = conn.cursor() # Создаём список custom_fields для хранения определений полей таблицы. custom_fields = [] # Проходимся по атрибутам класса object_type и извлекаем информацию о полях. for key, value in vars(self.object_type).items(): if not key.startswith("__") and not callable(value): field_name = key field_type = value.field_type is_unique = value.unique is_null = value.null default_value = value.default # Создаём строку с определением поля и добавляем её в список custom_fields. field_declaration = [f'"{field_name}" {field_type}'] if is_unique: field_declaration.append('UNIQUE') if not is_null: field_declaration.append('NOT NULL') if default_value is not None: field_declaration.append(f'DEFAULT {default_value}') custom_fields.append(' '.join(field_declaration)) # Создаём SQL-запрос для создания таблицы с определёнными полями. create_table_sql = f''' CREATE TABLE IF NOT EXISTS {self.object_type.__name__} ( {", ".join(custom_fields)} ); ''' # Выполняем SQL-запрос. cursor.execute(create_table_sql) # Фиксируем изменения и закрываем соединение с базой данных. conn.commit() conn.close()Создавать таблицу нужно для каждой пользовательской модели — для этого удобно использовать простой декоратор для класса:
def simple_orm(class_: type): EXTERN_TYPES[class_.__name__] = class_ # Сохраняем в словарь моделей class_.objects.__createTable__() # Создаём таблицу в бд return class_
Декораторы используются с помощью символа «@» перед определением функции, которую они декорируют. Вот пример использования декоратора:
def my_decorator(func): def wrapper(): print("Что-то происходит перед вызовом функции") func() print("Что-то происходит после вызова функции") return wrapper @my_decorator def say_hello(): print("Привет, мир!") say_hello()В этом примере my_decorator — это декоратор, который добавляет вывод текста до и после вызова функции say_hello. Затем декоратор @my_decorator применяется к функции say_hello, и при вызове say_hello() будет выполнено дополнительное действие, предусмотренное декоратором.
Декораторы часто используются для следующих задач:
- Логирования: Запись логов для функций или методов.
- Аутентификации: Проверка прав доступа перед вызовом функции.
- Кэширования: Сохранение результатов функции для ускорения будущих вызовов с теми же аргументами.
- Измерения времени выполнения: Оценка производительности функции.
- Модификации поведения: Изменение или расширение функциональности функции.
Python предоставляет множество встроенных декораторов, таких как @staticmethod, @classmethod, @property и другие, а также вы можете создавать свои собственные декораторы в соответствии с вашими потребностями.
Теперь при инициализации модели в бд создаётся таблица с соответствующими полями:
▍ JsonField
В sqlite3 нет JsonField по умолчанию, так что мы реализуем его на основе текста. Для начала добавляем его в базовые типы:
class JsonField(BaseType): field_type = 'JSON'Json, по сути, ведёт себя, как текст, за исключением того, что при создании и изменении надо использовать json.dumps(), а при получении — json.loads().
▍ ForeignKey
В реляционных базах данных, «foreign key» (внешний ключ) — это структурный элемент, который используется для установления связей между двумя таблицами. Внешний ключ представляет собой один или несколько столбцов в одной таблице, которые связаны с первичным ключом (обычно) в другой таблице. Эта связь позволяет базе данных поддерживать целостность данных и обеспечивать связи между данными в разных таблицах.
ForeignKey должен возвращать объект заданного типа по значению заданного поля.
class ForeignKey(BaseType): field_type = 'FOREIGN_KEY' def __init__(self, object_class: type, foreign_field: str, unique: bool = False, null: bool = True, default=None): self.object_class = object_class, self.foreign_field = foreign_field, self.unique = unique self.null = null self.default = defaultЯ реализую его просто, как json объект с параметрами type, key, value, например:
{"type": "Box", "key": "name", "value": "BOX 1"}.Теперь мы с помощью него можем делать так:
@simple_orm class Box(models.Model): name = TextField() width = IntegerField() height = IntegerField() @simple_orm class Circle(models.Model): box = ForeignKey(object_class=Box, foreign_field='name') name = TextField() radius = IntegerField() data = JsonField() box = Box.objects.add(Box('BOX 1', 1, 1)) circle = Circle.objects.add(Circle(box, "CIRCLE 1", 5, {'data': 5})) print(circle.json()) #{'box': <__main__.Box object at 0x7f7637f6d850>, 'name': 'CIRCLE 1', 'radius': 5, 'data': {'data': 5}}
▍ Добавляем CRUD
class Object: def __init__(self, object_type): self.object_type = object_type def add(self, obj): conn = sqlite3.connect(db_name) cursor = conn.cursor() d = copy.copy(obj.__dict__) object_type_name = self.object_type.__name__ for key, value in vars(self.object_type).items(): if not key.startswith("__") and not callable(value): if type(value) in BASIC_TYPES: continue if type(value) == JsonField: d[key] = json.dumps(d[key]) if type(value) == ForeignKey: d[key] = json.dumps({'type': value.object_class[0].__name__, 'key': value.foreign_field[0], 'value': getattr(d[key], value.foreign_field[0])}) insert_sql = f'INSERT INTO {object_type_name} ({", ".join(obj.__dict__.keys())}) VALUES ({", ".join(["?"] * len(obj.__dict__))});' values = tuple(d.values()) cursor.execute(insert_sql, values) conn.commit() conn.close() return obj def save(self, obj): conn = sqlite3.connect(db_name) cursor = conn.cursor() d = copy.copy(obj.__dict__) object_type_name = self.object_type.__name__ for key, value in vars(self.object_type).items(): if not key.startswith("__") and not callable(value): if type(value) in BASIC_TYPES: continue if type(value) == JsonField: d[key] = json.dumps(d[key]) if type(value) == ForeignKey: d[key] = json.dumps({'type': value.object_class[0].__name__, 'key': value.foreign_field[0], 'value': getattr(d[key], value.foreign_field[0])}) upsert_sql = f'INSERT OR REPLACE INTO {object_type_name} ({", ".join(obj.__dict__.keys())}) VALUES ({", ".join(["?"] * len(obj.__dict__))});' values = tuple(d.values()) cursor.execute(upsert_sql, values) conn.commit() conn.close() return obj def get(self, **kwargs): conn = sqlite3.connect(db_name) cursor = conn.cursor() object_type_name = self.object_type.__name__ attr_value_pairs = [(attr, value) for attr, value in kwargs.items()] where_clauses = [f'{attr} = ?' for attr, _ in attr_value_pairs] where_clause = ' AND '.join(where_clauses) select_by_attrs_sql = f'SELECT * FROM {object_type_name} WHERE {where_clause};' values = tuple(value for _, value in attr_value_pairs) cursor.execute(select_by_attrs_sql, values) row = cursor.fetchone() conn.close() if row: obj = self.object_type() for i, value in enumerate(row): if type(getattr(obj, cursor.description[i][0])) == JsonField: setattr(obj, cursor.description[i][0], json.loads(value)) elif type(getattr(obj, cursor.description[i][0])) == ForeignKey: value = json.loads(value) setattr(obj, cursor.description[i][0], EXTERN_TYPES[value['type']].objects.get(**{value['key']: value['value']})) else: setattr(obj, cursor.description[i][0], value) return obj else: return None def delete(self, **kwargs): conn = sqlite3.connect(db_name) cursor = conn.cursor() object_type_name = self.object_type.__name__ attr_value_pairs = [(attr, value) for attr, value in kwargs.items()] where_clauses = [f'{attr} = ?' for attr, _ in attr_value_pairs] where_clause = ' AND '.join(where_clauses) delete_by_attrs_sql = f'DELETE FROM {object_type_name} WHERE {where_clause};' values = tuple(value for _, value in attr_value_pairs) cursor.execute(delete_by_attrs_sql, values) conn.commit() conn.close() def filter(self, **kwargs): conn = sqlite3.connect(db_name) cursor = conn.cursor() object_type_name = self.object_type.__name__ attr_value_pairs = [(attr, value) for attr, value in kwargs.items()] where_clauses = [f'{attr} = ?' for attr, _ in attr_value_pairs] where_clause = ' AND '.join(where_clauses) select_by_attrs_sql = f'SELECT * FROM {object_type_name} WHERE {where_clause};' values = tuple(value for _, value in attr_value_pairs) cursor.execute(select_by_attrs_sql, values) rows = cursor.fetchall() conn.close() objects = [] for row in rows: obj = self.object_type() for i, value in enumerate(row): if type(getattr(obj, cursor.description[i][0])) == JsonField: setattr(obj, cursor.description[i][0], json.loads(value)) elif type(getattr(obj, cursor.description[i][0])) == ForeignKey: value = json.loads(value) setattr(obj, cursor.description[i][0], EXTERN_TYPES[value['type']].objects.get(**{value['key']: value['value']})) else: setattr(obj, cursor.description[i][0], value) objects.append(obj) return objects def all(self): conn = sqlite3.connect(db_name) cursor = conn.cursor() object_type_name = self.object_type.__name__ select_all_sql = f'SELECT * FROM {object_type_name};' cursor.execute(select_all_sql) rows = cursor.fetchall() conn.close() objects = [] for row in rows: obj = self.object_type() for i, value in enumerate(row): if type(getattr(obj, cursor.description[i][0])) == JsonField: setattr(obj, cursor.description[i][0], json.loads(value)) elif type(getattr(obj, cursor.description[i][0])) == ForeignKey: value = json.loads(value) setattr(obj, cursor.description[i][0], EXTERN_TYPES[value['type']].objects.get(**{value['key']: value['value']})) else: setattr(obj, cursor.description[i][0], value) objects.append(obj) return objects def __createTable__(self): conn = sqlite3.connect(db_name) cursor = conn.cursor() custom_fields = [] for key, value in vars(self.object_type).items(): if not key.startswith("__") and not callable(value): field_name = key field_type = value.field_type is_unique = value.unique is_null = value.null default_value = value.default if value.field_type == 'FOREIGN_KEY': field_type = "TEXT" if value.field_type == 'JSON': field_type = 'TEXT' field_declaration = [f'"{field_name}" {field_type}'] if is_unique: field_declaration.append('UNIQUE') if not is_null: field_declaration.append('NOT NULL') if default_value is not None: field_declaration.append(f'DEFAULT {default_value}') custom_fields.append(' '.join(field_declaration)) create_table_sql = f''' CREATE TABLE IF NOT EXISTS {self.object_type.__name__} ( {", ".join(custom_fields)} ); ''' cursor.execute(create_table_sql) conn.commit() conn.close()
▍ Список объектов
Сейчас функции all() и filter() возвращают list состоящий из объектов. Это неудобно, ведь нельзя, например, удалить все объекты. Исправим это, добавив класс ListOfObjects:
class ListOfObjects: def __init__(self, objects): self.objects = objects def filter(self, **kwargs): filtered_objects = [] for obj in self.objects: if all(getattr(obj, attr, None) == value for attr, value in kwargs.items()): filtered_objects.append(obj) return ListOfObjects(filtered_objects) def delete(self): for obj in self.objects: obj.delete() def json(self): object_dicts = [obj.json() for obj in self.objects] return object_dicts
▍ Примеры
models.py
import simple_orm.models as models from simple_orm.models import IntegerField, TextField, ForeignKey, JsonField from simple_orm.models import simple_orm from pathlib import Path import sys PATH = Path(__file__).absolute().parent.parent sys.path.append(str(PATH)) @simple_orm class Box(models.Model): name = TextField() width = IntegerField() height = IntegerField() @simple_orm class Circle(models.Model): box = ForeignKey(object_class=Box, foreign_field='name') name = TextField() radius = IntegerField() data = JsonField()main.py
from models import Box, Circle box = Box.objects.add(Box('BOX 1', 1, 1)) circle = Circle.objects.add(Circle(box, "CIRCLE 1", 5, {'data': 5})) print(circle.json()) print(Box.objects.filter(width=1, height=1).json()) print(Circle.objects.get(name="CIRCLE 1").json()) Box.objects.delete(name="BOX 1") print(Box.objects.all().json())
▍ Заключение
В процессе создания своего ORM мы использовали много сложных инструментов языка python, которые помогли написать короткий и красивый код, решающий довольно сложную задачу.
Да, он не идеальный, вы можете предложить его улучшения в комментариях.
Полный код есть на https://github.com/leo-need-more-coffee/simple-orm
Библиотека для python: https://pypi.org/project/sqlite3-simple-orm
Скачать с помощью pip: pip install sqlite3-simple-orm
Узнавайте о новых акциях и промокодах первыми из нашего Telegram-канала ?
ссылка на оригинал статьи https://habr.com/ru/articles/766552/
Добавить комментарий