Для асинхронного Python существует мало полноценных ORM, и им далеко до таких монстров-комбайнов, как DjangoOrm и SQLAlchemy.ORM. Бедность ORM-инструментария для асинхронного программирования заставила многих программистов отказаться от зачастую непонятной им работы с ORM и перейти к более прозрачному взаимодействию с БД. Решение в лоб — написание raw SQL, но в этом случае запросы не будут защищены от инъекций, а запросы, составляемые по бизнес логике с опциональными параметрами, превратятся в конкатенацию строк. Важно найти баланс между прозрачностью выполнения кода, скоростью его написания и читаемостью.
Ниже я предлагаю реализацию такого баланса c использованием SQLAlchemy Core.
Сравнение с остальными ORM
Прежде чем рассматривать работу SQLAlchemy Core, для расширения инженерного кругозора давайте познакомимся с альтернативными решениями и их недостатками.
Django ORM и ponyORM не подходят из-за невозможности работать в асинхронном режиме.
SQLAlchemy.ORM: поддержку полноценной асинхронной ORM в альфа-режиме добавили в версии 1.4, пока не подходит для production-решений.
Peewee: умеет работать в асинхронном режиме, но только с Core-функциями. Хорошая библиотека, но я предлагаю выбрать SQLAlchemy из-за её большей популярности и поддерживаемости. Также были жалобы на ленивую загрузку объектов в Peewee — иногда она намертво вешает цикл (возможно, эту ошибку уже поправили в новых версиях).
Gino: простая ORM на основе SQLAlchemy Core. Работать с этой библиотекой можно без знаний об SQLAlchemy.
tortoise-orm: молодой проект с похожим на Django ORM способом доступа к данным.
Работа с SQLAlchemy Core
Алхимия состоит из двух частей. Первая — это абстракция над SQL-базой данных, которая называется SQLAlchemy Сore. Вторая — это ORM, собственно mapping между реляционной БД и объектным представлением. Соответственно, SQLAlchemy Сore почти один к одному совпадает с SQL — если вы знаете последний, то проблем с Core, как правило, не возникает. Благодаря этому использование SQLAlchemy Сore имеет наименьший оверхэд при обращении к БД.
SQLAlchemy Core предоставляет всю основную функциональность для построения запросов. Но в крупном проекте при обращении к базовому API SQLAlchemy будет много дублирующихся участков, поэтому логично написать библиотеку-обёртку для реализации наиболее популярных сценариев использования.
Как видите, участки кода с передачей параметров повторяются, а сам код содержит много параметров, связанных с внутренней реализацией SQLAlchemy Core, что ухудшает читаемость. Хотелось бы получать данные из БД с помощью таких запросов:
user = await User.select().where(User.user_type == user_type).get(conn)
_ = await User.insert().values(val='abc').execute(conn)
Чтобы этого добиться, необходимо реализовать обработчики для операций insert, select, update и delete.
Пишем «обёртку»
Напишем для функции select полноценную обёртку, которая даст возможность с минимумом кода использовать все виды join, сортировки и остальные часто используемые операции:
class SelectQuery:
def __init__(self, model, fields):
self._model = model
self._from = None
_fields = []
self._relations = {field.class_ if isinstance(field, QueryableAttribute) else field for field in fields}
if model in self._relations:
self._relations.remove(model)
for field in fields:
if isinstance(field, QueryableAttribute):
_fields.append(getattr(field.class_.__table__.c, field.key))
elif isinstance(field, sa.Column):
_fields.append(field)
elif hasattr(field, '__table__'):
for f in field.__table__.c:
_fields.append(f)
else:
_fields.append(field)
self._stmt = sa.select([model, *self._relations]).with_only_columns(_fields)
def join(self, right, on):
if self._from is None:
self._from = sa.join(self._model, right, on)
else:
self._from = self._from.join(right, on)
return self
def outerjoin(self, right, on):
if self._from is None:
self._from = sa.outerjoin(self._model, right, on)
else:
self._from = self._from.outerjoin(right, on)
return self
def where(self, condition):
self._stmt = self._stmt.where(condition)
return self
def group_by(self, column):
self._stmt = self._stmt.group_by(column)
return self
def order_by(self, *fields):
self._stmt = self._stmt.order_by(*fields)
return self
def distinct(self, *fields):
self._stmt = self._stmt.distinct()
return self
def limit(self, limit):
self._stmt = self._stmt.limit(limit)
return self
def offset(self, offset):
self._stmt = self._stmt.offset(offset)
return self
async def get(self, conn):
if self._from is not None:
self._stmt = self._stmt.select_from(self._from)
result = await conn.execute(self._stmt)
result = await result.first()
return result
async def all(self, conn):
if self._from is not None:
self._stmt = self._stmt.select_from(self._from)
result = await conn.execute(self._stmt)
return await result.fetchall()
def get_query(self):
if self._from is not None:
return self._stmt.select_from(self._from)
return self._stmt
@property
def raw_sql(self):
stmt = self._stmt
if self._from is not None:
stmt = stmt.select_from(self._from)
return stmt.compile(compile_kwargs={'literal_binds': True})
def _create_model(self, model_class, data):
model_fields = {
getattr(field, 'name'): data[str(field)] for field in model_class.__table__.c if str(field) in data
}
model = model_class(**model_fields)
for relation in model_class.__mapper__.relationships:
if relation.mapper.class_ in self._relations:
relation_model = self._create_model(relation.mapper.class_, data)
setattr(model, relation.key, relation_model)
return model
Также специфическими операциями обладают запросы на удаление:
