fastapi-sqlalchemy-toolkit 0.6.1

FastAPI SQLAlchemy Toolkit

FastAPI SQLAlchemy Toolkit

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

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

• автоматическая валидация значений на уровне БД при создании и изменении объектов через API.

Для этого FastAPI SQLAlachemy Toolkit предоставляет класс менеджера fastapi_sqlalchemy_toolkit.ModelManager для взаимодействия с моделью.

Features

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

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

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

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

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

• Упрощение 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)

Атрибут 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 / paginated_filter - получение списка объектов с фильтрами и пагинацией через fastapi_pagination
• list / filter - получение списка объектов с фильтрами
• count - получение количества объектов
• update - обновление объекта; выполняет валидацию значений полей на уровне БД
• delete - удаление объекта
• bulk_create - создание объектов пачкой; выполняет валидацию значений полей на уровне БД
• bulk_update - обновление объектов пачкой; выполняет валидацию значений полей на уровне БД

Фильтрация

Для получения списка объектов с фильтрацией fastapi_sqlalchemy_toolkit предоставляет два метода: list, который имеет дополнительную предобработку значений, и filter, который не производит скрытых обработк. Аналогично ведут себя методы paginated_list и paginated_filter, за исключением того, что они пагинирует результат с помощью fastapi_pagination.

Пусть у нас есть следующие модели:

class Base(DeclarativeBase):
    id: Mapped[_py_uuid] = mapped_column(
        UUID(as_uuid=True), primary_key=True, default=uuid4
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )


class Parent(Base):
    title: Mapped[str]
    slug: Mapped[str] = mapped_column(unique=True)
    children: Mapped[list["Child"]] = relationship(back_populates="parent")


class Child(Base):
    title: Mapped[str]
    slug: Mapped[str] = mapped_column(unique=True)
    parent_id: Mapped[UUID] = mapped_column(ForeignKey("parent.id", ondelete="CASCADE"))
    parent: Mapped[Parent] = relationship(back_populates="children")

И менеджер:

from fastapi_sqlalchemy_toolkit import ModelManager

child_manager = ModelManager[Child, CreateChildSchema, PatchChildSchema](
    Child, default_ordering=Child.title
)

Простая фильтрация по точному соответствию через API

@router.get("/children")
async def get_list(
    session: Session,
    slug: str | None = None,
) -> list[ChildListSchema]:
    return await child_manager.list(
        session,
        slug=slug,
    )

Запрос GET /children сгенерирует следующий SQL код:

SELECT child.title, child.slug, child.parent_id, child.id, child.created_at 
FROM child

Запрос GET /children?slug=child-1 сгенерирует следующий SQL код:

SELECT child.title, child.slug, child.parent_id, child.id, child.created_at 
FROM child 
WHERE child.slug = :slug_1

По конвенции FastAPI, необязательные параметры запроса типизируются подобно slug: str | None = None. При этом ожидается, что при запросе GET /children будут возвращены все объекты Child, а не только те, у которых slug is null. Поэтому метод list (paginated_list) отбрасывает фильтрацию по этому параметру, если он не передан (читайте раздел "Фильтрация без дополнительной обработки" для того, чтобы узнать, как избежать такого поведения).

Более сложная фильтрация

Чтобы использовать фильтрацию не только по точному соответствию с атрибутом модели, в методах list и paginated_list можно использовать параметр filter_expressions.

Параметр filter_expressions принимает словарь, в котором ключи -- это:

1. Атрибуты основной модели (Child.title) или связанной модели (Parent.title)

2. Операторы атрибутов модели (Child.title.ilike)

3. Функции sqlalchemy над атрибутами модели (func.date(Child.created_at))

Значение по ключу -- это значение, по которому должна осуществляться фильтрация.

Пример фильтрации по оператору атрибута модели:

@router.get("/children")
async def get_list(
    session: Session,
    title: str | None = None,
) -> list[ChildListSchema]:
    return await child_manager.list(
        session,
        filter_expressions={
            Child.title.ilike: title
        },
    )

Запрос GET /children сгенерирует следующий SQL код:

SELECT child.title, child.slug, child.parent_id, child.id, child.created_at 
FROM child

Запрос GET /children?title=ch сгенерирует следующий SQL код:

SELECT child.title, child.slug, child.parent_id, child.id, child.created_at 
FROM child 
WHERE lower(child.title) LIKE lower(:title_1)

Пример фильтрации по атрибуту связанной модели:

@router.get("/children")
async def get_list(
    session: Session,
    parent_title: str | None = None,
) -> list[ChildListSchema]:
    return await child_manager.list(
        session,
        filter_expressions={
            Parent.title.ilike: title
        },
    )

Запрос GET /children?parent_title=ch сгенерирует следующий SQL код:

SELECT parent.title, parent.slug, parent.id, parent.created_at, 
child.title AS title_1, child.slug AS slug_1, child.parent_id, child.id AS id_1, child.created_at AS created_at_1 
FROM child LEFT OUTER JOIN parent ON parent.id = child.parent_id 
WHERE lower(parent.title) LIKE lower(:title_1)

При фильтрации по полям связанных моделей через параметр filter_expression необходимые для фильтрации join будут сделаны автоматически. Важно: работает только для моделей, напрямую связанных с основной.

Пример фильтрации по функции sqlalchemy над атрибутом модели:

@router.get("/children")
async def get_list(
    session: Session,
    created_at_date: date | None = None,
) -> list[ChildListSchema]:
    return await child_manager.list(
        session,
        filter_expressions={
            func.date(Child.created_at): created_at_date
        },
    )

Запрос GET /children?created_at_date=2023-11-19 сгенерирует следующий SQL код:

SELECT child.title, child.slug, child.parent_id, child.id, child.created_at 
FROM child 
WHERE date(child.created_at) = :date_1

Фильтрация без дополнительной обработки

Для фильтрации без дополнительной обработки в методах list и paginated_list можно использовать параметр where. Переданное в этот параметр значение будет напрямую передано в метод .where() экземпляра Select в выражении запроса SQLAlchemy.

    non_archived_items = await item_manager.list(session, where=(Item.archived_at == None))

Использовать параметр where методов list и paginated_list имеет смысл тогда, когда эти методы используются в списочном API эндпоинте и предобработка каких-то параметров запроса полезна, однако нужно также добавить фильтр без предобработок от fastapi_sqlalchemy_toolkit.

В том случае, когда предобработки fastapi_sqlalchemy_toolkit не нужны вообще, стоит использовать методы filter и paginated_filter:

    created_at = None

    items = await item_manager.filter(session, created_at=created_at)

SELECT item.id, item.name, item.created_at
FROM item
WHERE itme.created is null

Конкретно, в отличие от метода list, метод filter:

1. Не игнорирует простые фильтры (kwargs) со значением None

2. Не имеет параметра filter_expressions, т. е. не будет выполнять join, необходимые для фильтрации по полям связанных моделей.

Фильтрация по null через API

Если в списочном эндпоинет API требуется, чтобы можно было как отфильтровать значение поля по переданному значению, так и отфильтровать его по null, предлагается использовать параметр nullable_filter_expressions методов list (paginated_list):

from datetime import datetime

from fastapi_sqlalchemy_toolkit import NullableQuery

from app.managers import my_object_manager
from app.models import MyObject

@router.get("/my-objects")
async def get_my_objects(
    session: Session,
    deleted_at: datetime | NullableQuery | None = None
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        nullable_filter_expressions={
            MyObject.deleted_at: deleted_at
        }
    )

Параметру с поддержкой фильтрации по null нужно указать возможный тип fastapi_sqlalchemy_toolkit.NullableQuery.

Теперь при запросе GET /my-objects?deleted_at= или GET /my-objects?deleted_at=null вернутся объекты MyObject, у которых deleted_at IS NULL.

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

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

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

Предпосылки

Необязательный раздел с демо сокращения количества шаблонного кода при использовании fastapi_sqlalchemy_toolkit.

Если в эндпоинт 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()
Session = Annotated[AsyncSession, Depends(get_async_session)]


@router.get("/my-objects")
async def get_my_objects(
    session: Session,
    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=f"%{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 == f"%{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: Session,
    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,
        filter_expressions={
            MyObject.name: name,
            MyObjectParent.name: parent_name
        }
    )

Сортировка

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: Session,
    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: Session, 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: Session,
    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_"))