Салют Хабр! Меня зовут Влад и я хотел бы осветить довольно важную тему, которую я не видел в обучающих материалах, а именно сложные запросы через паттерн Repository.
Проблема: построение гибких SQLAlchemy запросов на основе данных, переданных из бизнес логики. Более масштабное использование паттерна Repository.
Решение(то, о чём говорится в этой статье): все запросы формируются посредством конфигов, переданных из бизнес логики, далее конфиги обрабатываются и из них составляется ORM запрос.
Схема работы реализации:
Configs, ORM Query Builder, Operator Agregate.Configs - это правила, по которым в бизнес логику должны вернуться данные.
Актуальность для читателя: я думаю, что эта статья способна показать работу с паттерном Repository под другим углом, показать как можно строить гибкие запросы без создания кучи методов в имплементациях и потом описывания этих же методов в интерфейсах.
Пожалуй, можно начинать.
Все вы видели те самые классы, которые мелькают в видео на Youtube.
from typing import Generic, TypeVar, Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
DTO = TypeVar("DTO")
class SQLAlchemyRepository(Generic[DTO]):
def __init__(self, session, model):
self._model = model
self._session = session
async def get_by_filter(self, filters: dict[str, Any]) -> DTO | None:
query = select(self._model).filter_by(**filters)
data = await self._session.execute(query)
data_scalar = data.scalar()
if data_scalar is None:
return None
return data_scalar.to_dto()
async def get_many_by_filter(self, filters: dict[str, Any]) -> list[DTO]:
query = select(self._model).filter_by(**filters)
data = await self._session.execute(query)
data_all = data.scalars().all()
if not data_all:
return []
return [sqal_model.to_dto() for sqal_model in data_all]При росте вашего проекта вам понадобятся более кастомизированные запросы в бд и логики этих методов хватать не будет, соответственно вам придётся расширяться.
Как уже упоминалось в Решение используются конфиги, думаю, сейчас лучший момент, чтобы их показать, но для начала нужно объявить структуру проекта.
repository/
builder_configs/
__init__.py
configs.py
types.pyРассмотрим содержимое файла configs.py
from dataclasses import dataclass, field
from typing import Any
from .types import FilterType, OrderByType, LazyLoadType, FilterLogicType, UNSET
@dataclass
class Filter:
column: str
value: Any
filter_type: FilterType = FilterType.EQ
@dataclass
class FilterConfig:
filters: list[Filter]
mode: FilterLogicType = FilterLogicType.NULL
@dataclass
class OrderByConfig:
column: str
mode: OrderByType = OrderByType.ASC
@dataclass
class ColumnConfig:
column: str
label: str | UNSET = UNSET
value: Any = UNSET
filter_type: FilterType = FilterType.EQ
value_is_column: bool = False
@dataclass
class JoinConfig:
table_name: str
columns: list[ColumnConfig] = field(default_factory=list)
filters: list[FilterConfig] = field(default_factory=list)
order_by: list[OrderByConfig] = field(default_factory=list)
@dataclass
class LazyLoadConfig:
relationship_strategy: str
load_type: LazyLoadType = LazyLoadType.JOINEDLOADХочу поподробней остановиться на конфигах.
-
Filter - простой в реализации dataclass, который принимает на вход поле, значение и необходимый оператор.
-
FilterConfig - в него можно передавать несколько фильтров и применять к ним логическое условие OR, AND. В случае если условие передано не будет (NULL), то все агрегированные объекты будут передавать в where позиционно.
-
OrderByConfig - сортировка поля по заданному режиму (ASC, DESC)
-
ColumnConfig - этот dataclass позволяет включать в SQL запрос выборку отдельных колонок, задавать им label, также он позволяет делать булевые выборки, также есть поддержка добаdления в булевую выборку полей из таблиц. Для этого нужно:
value=table_name.column,value_is_column=True. -
JoinConfig - позволяет присоединять другие таблицы. Переданные атрибуты
columns,filters,order_byбудут использованы для конкретной таблицыtable_name. -
LazyLoadConfig - поддержка подгрузки relationship. Чтобы подгрузить нужно указать название relationhip атрибута. Если же вы хотите получить вложенное отношение, то нужно указывать relationship по цепочке через
.-
Пример вложенных отношений:
Допустим, есть модель User и она имеет отношение с таблицей Item через атрибут
item, также Item связана с другой таблицей Collections посредством отношенияcollectionsи вот здесь и появляются вложенные отношения. Представим, что выполняется запрос в таблицу User и понадобилось подтянуть отношение item и к нему collections. В таком случаеrelationship_strategy=item.collections. Более подробно это будет рассмотрено при разборе реализации методаloadкласса SQLAlchemyQueryBuilder.
-
Теперь можно посмотреть на содержание файла types.py
from typing import NewType
from enum import Enum
UNSET = NewType("UNSET", None)
class FilterLogicType(Enum):
AND = "and"
OR = "or"
NULL = "null"
class FilterType(Enum):
EQ = "="
GT = ">"
GE = ">="
LT = "<"
LE = "<="
IN = "in"
VECTOR_SORT = "vector_sort"
class OrderByType(Enum):
ASC = "asc"
DESC = "desc"
class LazyLoadType(Enum):
JOINEDLOAD = "joinedload"
SELECTINLOAD = "selectinload"Всё, что находится в FilterType должен содержать Operator Agregate.
Теперь я хотел бы затронуть сам ORM Query Builder. Структура папок в свою очередь принимает такой вид:
repository/
builder_configs/
__init__.py
configs.py
types.py
query_builder/
__init__.py
interface.py
impl/
__init__.py
alchemy.pyИнтерфейс для Query Builder (query_builder/interface.py)
from typing import Protocol, Any
from typing_extensions import Self
from repository.builder_configs.configs import (
FilterConfig,
ColumnConfig,
JoinConfig,
LazyLoadConfig,
OrderByConfig
)
from .agregator.interface import AgregateFilterTypeProtocol
class QueryBuilderProtocol(Protocol):
def init(
self,
filter_agregate: AgregateFilterTypeProtocol,
*args,
**kwargs
) -> None:
self._filter_agregate = filter_agregate
def count(self) -> Self:
raise NotImplementedError
def columns(self, columns: list[ColumnConfig], *args, *kwargs) -> Self:
raise NotImplementedError
def filter(self, filters: list[FilterConfig], *args, *kwargs) -> Self:
raise NotImplementedError
def join(self, joins: list[JoinConfig]) -> Self:
raise NotImplementedError
def load(self, loads: list[LazyLoadConfig], *args, *kwargs) -> Self:
raise NotImplementedError
def order_by(self, order_by: list[OrderByConfig], *args, *kwargs) -> Self:
raise NotImplementedError
def values(self, data: dict[str, Any] | list[dict[str, Any]]) -> Self:
raise NotImplementedError
def limit(self, value: int | None) -> Self:
raise NotImplementedError
def offset(self, value: int | None) -> Self:
raise NotImplementedError
def build(self) -> Any:
raise NotImplementedErrorВ магическом методе __init__ вы можете заметить AgregateFilterTypeProtocol данный протокол нужен для агрегации операторов.
С появлением агрегатора структура папок примет такой вид:
repository/
builder_configs/
__init__.py
configs.py
types.py
query_builder/
__init__.py
interface.py
impl/
__init__.py
alchemy.py
agregator/
__init__.py
interface.py
impl/
__init__.py
alchemy.pyНиже вы можете увидеть интерфейс агрегатора (query_builder/agregator/interface.py)
from typing import Protocol, Any
from repository.builder_configs.types import FilterType
class AgregateFilterTypeProtocol(Protocol):
def filter_agregate(
self,
value: Any,
filter_type: FilterType,
*args,
**kwargs
) -> Any:
raise NotImplementedErrorРеализация агрегатора (query_builder/agregator/impl/alchemy.py)
from typing import Any
from sqlalchemy import func
from sqlalchemy.orm.properties import MappedColumn
from repository.builder_configs.types import FilterType
class SQLAlchemyAgregateFilterType:
def filter_agregate(
self,
value: Any,
filter_type: FilterType,
mapped_column: MappedColumn[Any],
*args,
**kwargs
) -> Any:
operator_with_lambda = {
"=": lambda column, value: column == value,
">": lambda column, value: column > value,
">=": lambda column, value: column >= value,
"<": lambda column, value: column < value,
"<=": lambda column, value: column <= value,
"in": lambda column, value: column.in_(value),
"vector_sort": lambda column, value: column.op("@@")(func.websearch_to_tsquery(str(value)))
}
if filter_type.value not in operator_with_lambda:
raise ValueError(f"Not found operator {filter_type.value}")
return operator_with_lambda[filter_type.value](mapped_column, value)В реализации вам может быть непонятен аргумент mapped_column - это экземпляр класса, полученный из атрибута ORM модели.
Ну и теперь гвоздь программы SQLAlchemyQueryBuilder. Я буду объяснять каждый метод постепенно.
from typing import Callable, Any, TypeVar
from typing_extensions import Self
from sqlalchemy.orm.properties import MappedColumn
from sqlalchemy.sql.elements import ColumnElement, Label, UnaryExpression
from sqlalchemy.orm.strategy_options import _AbstractLoad
from sqlalchemy import or_, and_, func, asc, desc
from sqlalchemy.orm import MappedColumn, joinedload, selectinload
from db.models import Base
from repository.builder_configs.configs import (
FilterConfig,
ColumnConfig,
JoinConfig,
OrderByConfig,
LazyLoadConfig
)
from repository.builder_configs.types import LazyLoadType, OrderByType, FilterLogicType, UNSET
from repository.query_builder.agregator.impl.alchemy import SQLAlchemyAgregateFilterType
ALCHEMYMODEL = TypeVar("ALCHEMYMODEL", bound=Base)
class SQLAlchemyQueryBuilder:
def __init__(
self,
query_type: Callable,
model: type[ALCHEMYMODEL],
filter_agregate: SQLAlchemyAgregateFilterType,
*args,
**kwargs
) -> None:
self._query_type: Callable = query_type
self._filter_agregate: SQLAlchemyAgregateFilterType = filter_agregate
self._model: type[ALCHEMYMODEL] = model
self._columns: list[MappedColumn[Any] | ColumnElement[bool] | Label[Any]] = []
self._limit: int | None = None
self._offset: int | None = None
self._joins: list[type[ALCHEMYMODEL]] = []
self._filter: list[ColumnElement[bool]] = []
self._order_by: list[UnaryExpression] = []
self._values: list[dict[str, Any]] = []
self._loads: list[_AbstractLoad] = []
self._count: bool = False
self._orm_models: dict[str, type["Base"]] = model.orm_models()query_type - тип выполняемого запроса: select, update и тп.
model - orm модель
После создания экземпляра этого класса и при вызове методов данные накапливаются в атрибутах и при вызове метода build строится ORM запрос.
Здесь можно обратить внимание на атрибут _orm_models, исходя из его аннотации можно сказать, что он хранит все зарегистрированные sqlalchemy модели.
Реализация метода orm_models
class Base(DeclarativeBase):
_cached_orm_models: dict[str, type["Base"]] = {}
@classmethod
def orm_models(cls) -> dict[str, type["Base"]]:
if not cls._cached_orm_models:
models = {}
for mapper in cls.registry.mappers:
models[mapper.class_.__tablename__] = mapper.class_
cls._cached_orm_models.update(models)
return cls._cached_orm_modelsТеперь можно перейти к самим методам класса SQLAlchemyQueryBuilder.
count
def count(self) -> Self:
self._count = True
return selfПозволяет получать количество записей.
columns
def columns(
self,
columns: list[ColumnConfig] = [],
model: type[ALCHEMYMODEL] | None = None,
*args,
**kwargs
) -> Self:
if columns:
if model is None:
model = self._model
for config in columns:
obj = getattr(model, config.column, None)
if (obj is None):
raise ValueError(f"model {model.__name__} has not column {config.column}")
if config.value is not UNSET:
if config.value_is_column is True:
table_name, column_name = config.value.split(".")
orm_object = self._orm_models.get(table_name)
if (hasattr(orm_object, column_name) is False):
raise ValueError(f"model {orm_object} has not column {column_name}")
config.value = getattr(orm_object, column_name)
obj = self._filter_agregate.filter_agregate(
mapped_column=obj,
value=config.value,
filter_type=config.filter_type
)
if config.label is not None:
obj = obj.label(config.label)
self._columns.append(obj)
return selfВ начале функции проверяется, была ли передана модель в функцию, это нужно, чтобы переиспользовать метод для разных моделей. Дальше - итерация, в которой из объекта model достаётся атрибут config.column. После идёт проверка на переданное value и если оно является полем в другой таблице, то строка распиливается по . и из _orm_models достаётся нужная orm модель, после в config.value записывается экземпляр класса MappedColumn полученный из атрибута column_name. Дальше работает агрегатор и как вы можете заметить в config.value может лежать как произвольное значение так и MappedColumn, далее это добавление label ну и окончательное действие - запись получившегося объекта в список _columns.
filter
def filter(
self,
filters: list[FilterConfig] = [],
model: type[ALCHEMYMODEL] | None = None,
*args,
**kwargs
) -> Self:
if filters:
if model is None:
model = self._model
configs = []
for filter in filters:
mode = None
if filter.mode == FilterLogicType.OR:
mode = or_
elif filter.mode == FilterLogicType.AND:
mode = and_
elements_objects = []
for filter_ in filter.filters:
obj = getattr(model, filter_.column, None)
if (obj is None):
raise ValueError(f"model {model.__name__} has not column {filter_.column}")
column_element = self._filter_agregate.filter_agregate(
mapped_column=obj,
filter_type=filter_.filter_type,
value=filter_.value
)
elements_objects.append(column_element)
if mode is None:
configs.extend(elements_objects)
else:
configs.append(mode(*elements_objects))
if configs:
self._filter.extend(configs)
return selfИтерация по FilterConfig, определение логического условия, после чего происходит итерация по Filter, в этом блоке кода также достаётся MappedColumn и отправляется в агрегатор. В element_objects остаются все аргегированные Filter, в configs - FilterConfig. После завершения основного цикла изменения отправляются в атрибут _filters.
order_by
def order_by(
self,
order_by: list[OrderByConfig] = [],
model: type[ALCHEMYMODEL] | None = None,
*args,
**kwargs
) -> Self:
if order_by:
if model is None:
model = self._model
for oby in order_by:
oby_mode = asc
if oby.mode == OrderByType.DESC:
oby_mode = desc
obj = getattr(model, oby.column, None)
if (obj is None):
raise ValueError(f"model {model.__name__} has not column {oby.column}")
self._order_by.append(oby_mode(obj))
return selfЯ думаю, что здесь объяснения излишни так как используются всё те же принципы, что и в предыдущих методах.
join
def join(self, joins: list[JoinConfig]) -> Self:
if joins:
for join in joins:
orm_object = self._orm_models.get(join.table_name)
if join.columns:
self.columns(
columns=join.columns,
model=orm_object
)
if join.filters:
self.filter(
filters=join.filters,
model=orm_object
)
if join.order_by:
self.order_by(
order_by=join.order_by,
model=orm_object
)
self._joins.append(orm_object)
return selfЗдесь можно заметить, что переиспользуются методы filter, columns, order_by, только
вместо изначально переданной в init orm модели, будет использована модель, название
которой указано в конфиге, объект этой модели достаётся из _orm_models.
load
def load(
self,
loads: list[LazyLoadConfig],
model: type[Base] | None = None,
*args,
**kwargs
) -> Self:
if loads:
for load in loads:
load_mode = joinedload
if load.load_type == LazyLoadType.SELECTINLOAD:
load_mode = selectinload
current_model = model
if model is None:
current_model = self._model
current_load = None
relationship_path = load.relationship_strategy.split(".")
for part in relationship_path:
relationship_declared = getattr(current_model, part, None)
if (relationship_declared is None):
raise ValueError(f"Error in relationship path: model {current_model.__name__} has no relationship {part}")
current_model = relationship_declared.mapper.class_
if current_load is None:
current_load = load_mode(relationship_declared)
else:
current_load = getattr(current_load, load_mode.__name__)(relationship_declared)
if current_load:
self._loads.append(current_load)
return selfЗдесь я бы хотел объяснить как именно обрабатывается relationship_strategy и как собираются вложенные relationship. Переменная relationship_path хранит все отношения, из которых составляется цепочка, итерируясь по ним, relationship_declared получает экземпляр класса _RelationshipDeclared и после current_model перезаписывается на модель, с которой связано отношение. После чего формируется current_load, если оно не было до этого записано, то просто передаём отношение в функцию load_mode, иначе мы вызываем функцию load_mode у current_load. Пример: joinedload(part1_relationship).joinedload(part2_relationship). Именно так и строятся вложенные relationship.
Теперь покажу несколько легковесных методов, в которых я думаю вы уже можете разобраться самостоятельно.
values, limit, offset
def values(
self,
data: dict[str, Any] | list[dict[str, Any]]
) -> Self:
if isinstance(data, dict):
self._values.append(data)
elif isinstance(data, list):
self._values.extend(data)
return self
def limit(
self,
value: int | None = None
) -> Self:
if value:
self._limit = value
return self
def offset(
self,
value: int | None = None
) -> Self:
if value:
self._offset = value
return selfbuild
def build(
self
) -> Any:
query = self._query_type(self._model)
if self._columns:
query = self._query_type(*self._columns).select_from(self._model)
if self._count:
count = func.count()
if self._columns:
count = func.count(*self._columns)
query = self._query_type(count).select_from(self._model)
if self._values:
query = query.values(self._values)
if self._filter:
query = query.filter(*self._filter)
if self._joins:
for join in self._joins:
query = query.join(join)
if self._order_by:
query = query.order_by(*self._order_by)
if self._loads:
query = query.options(*self._loads)
if self._limit:
query = query.limit(self._limit)
if self._offset:
query = query.offset(self._offset)
return queryДанный метод собирает ORM запрос.
Больше сказать мне нечего, всё показал, всё рассказал. Надеюсь, что эта статья даст читающим новые мысли по поводу паттерна Repository и то как с ним можно работать. Очень надеюсь на конструктивную критику и буду рад, если вы предложите свои реализации решения данной проблемы.
В этом репозитории вы можете посмотреть примеры использования.
Спасибо тебе, дорогой читатель, за интерес к моей статье!
Автор: shayzi
