fastapi-sqlalchemy-toolkit 0.3.0

FastAPI SQLAlchemy Toolkit

FastAPI SQLAlchemy Toolkit

FastAPI SQLAlchemy Toolkit — это библиотека для стека FastAPI + Async SQLAlchemy, которая помогает решать следующие задачи:

• cнижение количества шаблонного, копипастного кода, который возникает при разработке REST API и взаимодействии с СУБД через SQLAlchemy;

• валидация значений на уровне БД.

Features

• Декларативная фильтрация с помощью FieldFilter, в том числе по полям связанных моделей (см. раздел Фильтрация)

• Декларативная сортировка с помощью ordering_dep, в том числе по полям связанных моделей (см. раздел Сортировка)

• Методы для CRUD-операций с объектами в БД

• Валидация существования внешних ключей

• Валидация уникальных ограничений

• Упрощение CRUD-действий с M2M связями

Установка

pip install fastapi-sqlalchemy-toolkit

Quick Start

Пример использования fastapi-sqlalchemy-toolkit доступен в директории examples/app

Инициализация ModelManager

Для использования fastapi-sqlaclhemy-toolkit необходимо создать экземпляр ModelManager для своей модели:

from fastapi_sqlalchemy_toolkit import ModelManager

from .models import MyModel
from .schemas import MyModelCreateSchema, MyModelUpdateSchema

my_model_manager = ModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](MyModel)

При инициализации ModelManager можно задать параметр fk_mapping, необходимый для валидации внешних ключей. fk_mapping — это словарь, в котором ключи — это названия внешних ключей, а значения — модели SQLAlchemy, на которые эти ключи ссылаются.

from fastapi_sqlalchemy_toolkit import ModelManager

from .models import MyModel, MyParentModel
from .schemas import MyModelCreateSchema, MyModelUpdateSchema

my_model_manager = ModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](
    MyModel, fk_mapping={"parent_id": MyParentModel}
)

Атрибут default_ordering определяет сортировку по умолчанию при получении списка объектов. В него нужно передать поле основной модели.

from fastapi_sqlalchemy_toolkit import ModelManager

from .models import MyModel
from .schemas import MyModelCreateSchema, MyModelUpdateSchema

my_model_manager = ModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](
    MyModel, default_ordering=MyModel.title
)

Доступные методы ModelManager

Ниже перечислены CRUD методы, предоставляемые ModelManager. Документация параметров, принимаемых методами, находится в докстрингах методов.

• create - создание объекта; выполняет валидацию значений полей на уровне БД
• get - получение объекта
• get_or_404 - получение объекта или ошибки HTTP 404
• exists - проверка существования объекта
• paginated_list - получение списка объектов с фильтрами и пагинацией через fastapi_pagination
• list - получение списка объектов с фильтрами
• count - получение количества объектов
• update - обновление объекта; выполняет валидацию значений полей на уровне БД
• delete - удаление объекта
• bulk_create - создание объектов пачкой; выполняет валидацию значений полей на уровне БД
• bulk_update - обновление объектов пачкой; выполняет валидацию значений полей на уровне БД

Фильтрация

Предпосылки

Если в эндпоинт FastAPI нужно добавить фильтры по значениям полей, то код будет выглядеть примерно так:

from typing import Annotated
from uuid import UUID

from fastapi import APIRouter, Depends, Response, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.deps import get_async_session
from app.models import MyModel, MyParentModel
from app.schemas import MyObjectListSchema

router = APIRouter()
CurrentSession = Annotated[AsyncSession, Depends(get_async_session)]


@router.get("/my-objects")
async def get_my_objects(
    session: CurrentSession,
    user_id: UUID | None = None,
    name: str | None = None,
    parent_name: str | None = None,
) -> list[MyObjectListSchema]:
    stmt = select(MyModel)
    if user_id is not None:
        stmt = stmt.filter_by(user_id=user_id)
    if name is not None:
        stmt = stmt.filter(MyModel.name.ilike == name)
    if parent_name is not None:
        stmt = stmt.join(MyModel.parent)
        stmt = stmt.filter(ParentModel.name.ilike == parent_name)
    result = await session.execute(stmt)
    return result.scalars().all()

Как можно заметить, для реализации фильтрации необходима дубликация шаблонного кода.

В fastapi-sqlalchemy-toolkit этот эндпоинт выглядит так:

from fastapi_sqlalchemy_toolkit import FieldFilter

from app.managers import my_object_manager

@router.get("/my-objects")
async def get_my_objects(
    session: CurrentSession,
    user_id: UUID | None = None,
    name: str | None = None,
    parent_name: str | None = None,
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        user_id=user_id,
        name=FieldFilter(name, operator="ilike"),
        parent_name=FieldFilter(parent_name, operator="ilike", model=ParentModel, alias="name"),
    )

Использование FieldFilter

Дополнительные возможности декларативной фильтрации поддерживаются использованием класса FieldFilter. FieldFilter позволяет:

• фильтровать по значениям полей связанных моделей при установке атрибута model. При этом ModelManager автоматически сделает необходимые join'ы, если это модель, которая напрямую связана с главной
• использовать любые методы и атрибуты полей SQLAlchemy через атрибут operator
• применять функции SQLAlchemy к полям (например, date()) через атрибут func

from app.managers import parent_manager
from app.models import Child

from fastapi_sqlalchemy_toolkit import FieldFilter

await parent_manager.list(
    session, title=FieldFilter(child_title, model=Child, operator="ilike")
)

Фильтрация по null и необязательные квери параметры

Рассмотрим три сценария:

1. Эндпоинт с необязательными квери параметрами:

from fastapi_sqlalchemy_toolkit import FieldFilter

from app.managers import my_object_manager

@router.get("/my-objects")
async def get_my_objects(
    session: CurrentSession,
    user_id: UUID | None = None,
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        user_id=user_id,
    )

При запросе GET /my-objects без квери параметров вернутся все объекты MyObject, т. е. фильтр по user_id не будет применён. Аналогично с методом paginated_list.

2. Эндпоинт с возможностью фильтрации по null

from fastapi_sqlalchemy_toolkit import FieldFilter

from app.managers import my_object_manager

@router.get("/my-objects")
async def get_my_objects(
    session: CurrentSession,
    user_id: UUID | None = None,
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        user_id=user_id,
        filter_by_null=True
    )

При запросе GET /my-objects без квери параметров вернутся объекты MyObject, у которых user_id IS NULL. Аналогично с методом paginated_list.

Для того чтобы только часть квери параметров вызывала фильтрацию по null, можно использовать параметр filter_by_null класса FieldFilter:

from fastapi_sqlalchemy_toolkit import FieldFilter

from app.managers import my_object_manager

@router.get("/my-objects")
async def get_my_objects(
    session: CurrentSession,
    user_id: UUID | None = None,
    title: str | None = None
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        user_id=FieldFilter(user_id, filter_by_null=True),
        title=title
    )

При запросе GET /my-objects без квери параметров вернутся объекты MyObject, у которых user_id IS NULL, а фильтр по title не будет применён. Аналогично с методом paginated_list.

3. Фильтрация по null при использовании ModeManager как ORM

Если в методах ModelManager нужно получить список объектов с ожидаемым поведением фильтрации, где:

not_deleted_objects = await my_object_manager.list(
    session,
    deleted_at=None,
    filter_by_null=True
)

будет возвращать те объекты, у которых filter_by IS NULL, то нужно использовать параметр filter_by_null=True.

Альтернативно предлагается использовать метод .filter(), который аналогичен методу .list() с той разницей, что метод .filter() всегда имеет filter_by_null=True.

Фильтрация по обратным связям

Также в методах list и paginated_list есть поддержка фильтрации по обратным связям (relationship() в направлении один ко многим) с использованием метода .any().

# Если ParentModel.children -- это связь один ко многим
await parent_manager.list(session, children=[1, 2])
# Вернёт объекты Parent, у которых есть связь с ChildModel с id 1 или 2

Сортировка

fastapi-sqlalchemy-toolkit поддеживает декларативную сортировку по полям модели, а также по полям связанных моделей. При этом необходимые для сортировки по полям связанных моделей join'ы будут сделаны автоматически.

Для применения декларативной сортировки нужно:

1. Определить список полей, по которым доступна фильтрация. Поле может быть строкой, если это поле основной модели, или атрибутом модели, если оно находится на связанной модели.

from app.models import Parent

child_ordering_fields = (
    "title",
    "created_at",
    Parent.title,
    Parent.created_at
)

Для каждого из указаных полей будет доступна сортировка по возрастанию и убыванию. Чтобы сортировать по полю по убыванию, нужно в квери параметре сортировки передать его название, начиная с дефиса (Django style). Таким образом, ?order_by=title сортирует по title по возрастанию, а ?order_by=-title сортирует по title по убыванию.

2. В параметрах энпдоинта передать определённый выше список в ordering_dep

from fastapi_sqlalchemy_toolkit import ordering_dep

@router.get("/children")
async def get_child_objects(
    session: CurrentSession,
    order_by: ordering_dep(child_ordering_fields)
) -> list[ChildListSchema]
    ...

3. Передать параметр сортировки как параметр order_by в методы ModelManager

    return await child_manager.list(session=session, order_by=order_by)

Расширение

Методы ModelManager легко расширить дополнительной логикой.

В первую очередь необходимо определить свой класс ModelManager:

from fastapi_sqlalchemy_toolkit import ModelManager


class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager):
    ...

Дополнительная валидация

Дополнительную валидацию можно добавить, переопределив метод validate:

class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager):
    async def validate_parent_type(self, session: AsyncSession, validated_data: ModelDict) -> None:
        """
        Проверяет тип выбранного объекта Parent
        """
        # объект Parent с таким ID точно есть, так как это проверяется ранее в super().validate
        parent = await parent_manager.get(session, id=in_obj["parent_id"])
        if parent.type != ParentTypes.CanHaveChildren:
            raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="This parent has incompatible type",
                )
    
    async def run_db_validation(
            self,
            session: AsyncSession,
            db_obj: MyModel | None = None,
            in_obj: ModelDict | None = None,
        ) -> ModelDict:
        validated_data = await super().validate(session, db_obj, in_obj)
        await self.validate_parent_type(session, validated_data)
        return validated_data

Дополнительная бизнес логика при CRUD операциях

Если при CRUD операциях с моделью необходимо выполнить какую-то дополнительную бизнес логику, это можно сделать, переопределив соответствующие методы ModelManager:

class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager):
    async def create(
        self, *args, background_tasks: BackgroundTasks | None = None, **kwargs
    ) -> MyModel:
    created = await super().create(*args, **kwargs)
    background_tasks.add_task(send_email, created.id)
    return created

Такой подход соответствует принципу "Fat Models, Skinny Views" из Django.

Использование декларативных фильтров в нестандартных списочных запросах

Если необходимо получить не просто список объектов, но и какие-то другие поля (допустим, кол-во дочерних объектов) или агрегации, но также необходима декларативная фильтрация, то можно новый свой метод менеджера, вызвав в нём метод super().get_filter_expression:

class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](MyModel):
    async def get_parents_with_children_count(
        self, session: AsyncSession, **kwargs
    ) -> list[RetrieveParentWithChildrenCountSchema]:
        children_count_query = (
            select(func.count(Child.id))
            .filter(Child.parent_id == Parent.id)
            .scalar_subquery()
        )
        query = (
            select(Parent, children_count_query.label("children_count"))
        )

        # Вызываем метод для получения фильтров SQLAlchemy из аргументов методов
        # list и paginated_list
        query = query.filter(self.get_filter_expression(**kwargs))

        result = await session.execute(query)
        result = result.unique().all()
        for i, row in enumerate(result):
            row.Parent.children_count = row.children_count
            result[i] = row.Parent
        return result

Другие полезности

Сохранение пользователя запроса

Пользователя запроса можно задать в создаваемом/обновляемом объекте, передав дополнительный параметр в метод create (update):

@router.post("")
async def create_child(
    child_in: CreateUpdateChildSchema, session: CurrentSession, user: CurrentUser
) -> CreateUpdateChildSchema:
    return await child_manager.create(session=session, in_obj=child_in, author_id=user.id)

Создание и обновление объектов с M2M связями

Если на модели определена M2M связь, то использование ModelManager позволяет передать в это поле список ID объектов.

fastapi-sqlalchemy-toolkit провалидирует существование этих объектов и установит им M2M связь, без необходимости создавать отдельные эндпоинты для работы с M2M связями.

# Пусть модели Person и House имеют M2M связь
from pydantic import BaseModel


class PersonCreateSchema(BaseModel):
    house_ids: list[int]

...

    in_obj = PersonCreateSchema(house_ids=[1, 2, 3])
    await person_manager.create(session, in_obj)
    # Создаст объект Person и установит ему M2M связь с House с id 1, 2 и 3

Фильтрация по списку значений

Один из способов фильтрации по списку значений -- передать этот список в качестве квери параметра в строку через запятую. fastapi-sqlalchemy-toolkit предоставляет утилиту для фильтрации по списку значений, переданного в строку через запятую:

from uuid import UUID
from fastapi_sqlalchemy_toolkit.utils import comma_list_query, get_comma_list_values

@router.get("/children")
async def get_child_objects(
    session: CurrentSession,
    ids: comma_list_query = None,
) -> list[ChildListSchema]
    ids = get_comma_list_values(ids, UUID)
    return await child_manager.list(session, id=FieldFilter(ids, operator="in_"))