CRUD

Generic async CRUD operations for SQLAlchemy models with search, pagination, and many-to-many support. This module has features that are only compatible with Postgres.

!!! info This module has been coded and tested to be compatible with PostgreSQL only.

Overview

The crud module provides AsyncCrud, an abstract base class with a full suite of async database operations, and CrudFactory, a convenience function to instantiate it for a given model.

Creating a CRUD class

from fastapi_toolsets.crud import CrudFactory
from myapp.models import User

UserCrud = CrudFactory(model=User)

CrudFactory dynamically creates a class named AsyncUserCrud with User as its model.

Basic operations

# Create
user = await UserCrud.create(session=session, obj=UserCreateSchema(username="alice"))

# Get one (raises NotFoundError if not found)
user = await UserCrud.get(session=session, filters=[User.id == user_id])

# Get first or None
user = await UserCrud.first(session=session, filters=[User.email == email])

# Get multiple
users = await UserCrud.get_multi(session=session, filters=[User.is_active == True])

# Update
user = await UserCrud.update(session=session, obj=UserUpdateSchema(username="bob"), filters=[User.id == user_id])

# Delete
await UserCrud.delete(session=session, filters=[User.id == user_id])

# Count / exists
count = await UserCrud.count(session=session, filters=[User.is_active == True])
exists = await UserCrud.exists(session=session, filters=[User.email == email])

Pagination

@router.get(
    "",
    response_model=PaginatedResponse[User],
)
async def get_users(
    session: SessionDep,
    items_per_page: int = 50,
    page: int = 1,
):
    return await crud.UserCrud.paginate(
        session=session,
        items_per_page=items_per_page,
        page=page,
    )

The paginate function will return a PaginatedResponse.

Search

Declare searchable fields on the CRUD class. Relationship traversal is supported via tuples:

PostCrud = CrudFactory(
    model=Post,
    searchable_fields=[
        Post.title,
        Post.content,
        (Post.author, User.username),  # search across relationship
    ],
)

This allow to do a search with the paginate function:

@router.get(
    "",
    response_model=PaginatedResponse[User],
)
async def get_users(
    session: SessionDep,
    items_per_page: int = 50,
    page: int = 1,
    search: str | None = None,
):
    return await crud.UserCrud.paginate(
        session=session,
        items_per_page=items_per_page,
        page=page,
        search=search,
    )

Relationship loading

!!! info "Added in v1.1"

By default, SQLAlchemy relationships are not loaded unless explicitly requested. Instead of using lazy="selectin" on model definitions (which is implicit and applies globally), define a default_load_options on the CRUD class to control loading strategy explicitly.

!!! warning Avoid using lazy="selectin" on model relationships. It fires silently on every query, cannot be disabled per-call, and can cause unexpected cascading loads through deep relationship chains. Use default_load_options instead.

from sqlalchemy.orm import selectinload

ArticleCrud = CrudFactory(
    model=Article,
    default_load_options=[
        selectinload(Article.category),
        selectinload(Article.tags),
    ],
)

default_load_options applies automatically to all read operations (get, first, get_multi, paginate). When load_options is passed at call-site, it fully replaces default_load_options for that query — giving you precise per-call control:

# Only loads category, tags are not loaded
article = await ArticleCrud.get(
    session=session,
    filters=[Article.id == article_id],
    load_options=[selectinload(Article.category)],
)

# Loads nothing — useful for write-then-refresh flows or lightweight checks
articles = await ArticleCrud.get_multi(session=session, load_options=[])

Many-to-many relationships

Use m2m_fields to map schema fields containing lists of IDs to SQLAlchemy relationships. The CRUD class resolves and validates all IDs before persisting:

PostCrud = CrudFactory(
    model=Post,
    m2m_fields={"tag_ids": Post.tags},
)

post = await PostCrud.create(session=session, obj=PostCreateSchema(title="Hello", tag_ids=[1, 2, 3]))

Upsert

Atomic INSERT ... ON CONFLICT DO UPDATE using PostgreSQL:

await UserCrud.upsert(
    session=session,
    obj=UserCreateSchema(email="alice@example.com", username="alice"),
    index_elements=[User.email],
    set_={"username"},
)

`schema` — typed response serialization

!!! info "Added in v1.1"

Pass a Pydantic schema class to create, get, update, or paginate to serialize the result directly into that schema and wrap it in a Response[schema] or PaginatedResponse[schema]:

class UserRead(PydanticBase):
    id: UUID
    username: str

@router.get(
    "/{uuid}",
    responses=generate_error_responses(NotFoundError),
)
async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]:
    return await crud.UserCrud.get(
        session=session,
        filters=[User.id == uuid],
        schema=UserRead,
    )

@router.get("")
async def list_users(session: SessionDep, page: int = 1) -> PaginatedResponse[UserRead]:
    return await crud.UserCrud.paginate(
        session=session,
        page=page,
        schema=UserRead,
    )

The schema must have from_attributes=True (or inherit from PydanticBase) so it can be built from SQLAlchemy model instances.

!!! warning "Deprecated: as_response" The as_response=True parameter is deprecated and will be removed in v2.0. Replace it with schema=YourSchema.

:material-api: API Reference

6.4 KiB Raw Blame History