Skip to main content

FastAPI SQLAlchemy Toolkit

Project description

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 и необязательные квери параметры

Для получения списка объектов в fastapi-sqlalchemy-toolkit могут использоваться два метода: list и filter. Их отличие в том, что list игнорирует параметры фильтрации со значением None, в то время как filter применяет параметры фильтрации со значением None как SQL фильтр по NULL.

Таким образом, метод list подходит для использования в API эндпоинтах, где зачастую нужны необязательные квери параметры. Метод filter подходит для использования в качестве аналогичного метода ORM SQLAlchemy (в методах менеджеров и т. п.).

Аналогичным образом работают методы paginated_list и paginated_filter.

Примечание: метода list был добавлен, чтобы сохранить конвенции FastAPI, когда необязательные квери параметры объявляются как q: type_ | None = None. При этом ожидается, что при отсутствии параметра фильтр не будет применён.

Примеры:

  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,
    title: str | None = None
) -> list[MyObjectListSchema]:
    return await my_object_manager.list(
        session,
        user_id=user_id,
        title=FieldFilter(title, operator="icontains")
    )

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

Аналогично работает с методом paginated_list.

  1. Фильтрация по null при использовании ModelManager как ORM:
    async def my_manager_method(self, session):
        not_deleted_objects = await my_object_manager.filter(
            session,
            deleted_at=None,
        )

        ...

Использование метода filter обеспечивает, что note_deleted_objects содержит те объекты, у которых "deleted_at" IS NULL.

Аналогично работает с методом paginated_filter.

  1. Эндпоинт с возможностью фильтрации по null через квери параметры:
from datetime import datetime

from fastapi_sqlalchemy_toolkit import FieldFilter, NullableQuery

from app.managers import my_object_manager

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

Если требуется фильтрация значения поля по null через квери параметр, то необходимо указать допустимый тип соответствующего аргумента как fastapi_sqlalchemy_toolkit.NullableQuery и передать его в метод фильтрации с помощью FieldFilter с указанным параметрам nullable_q=True.

NullableQuery -- это пустая строка "" или строка "null"

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

Аналогично работает с методом 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 по убыванию.

  1. В параметрах энпдоинта передать определённый выше список в 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]
    ...
  1. Передать параметр сортировки как параметр 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_"))

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fastapi_sqlalchemy_toolkit-0.3.4.tar.gz (18.7 kB view hashes)

Uploaded Source

Built Distribution

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page