fastapi-toolsets/docs/module/crud.md

CRUD

Generic async CRUD operations for SQLAlchemy models with search, pagination, and many-to-many support.

!!! 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

!!! info "Added in v1.1 (only offset_pagination via paginate if <v1.1)"

Two pagination strategies are available. Both return a PaginatedResponse but differ in how they navigate through results.

	offset_paginate	cursor_paginate
Total count	Yes	No
Jump to arbitrary page	Yes	No
Performance on deep pages	Degrades	Constant
Stable under concurrent inserts	No	Yes
Search compatible	Yes	Yes
Use case	Admin panels, numbered pagination	Feeds, APIs, infinite scroll

Offset 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.offset_paginate(
        session=session,
        items_per_page=items_per_page,
        page=page,
    )

The offset_paginate method returns a PaginatedResponse whose pagination field is an OffsetPagination object:

{
  "status": "SUCCESS",
  "data": ["..."],
  "pagination": {
    "total_count": 100,
    "page": 1,
    "items_per_page": 20,
    "has_more": true
  }
}

!!! warning "Deprecated: paginate" The paginate function is a backward-compatible alias for offset_paginate. This function is deprecated and will be removed in v2.0.

Cursor pagination

@router.get(
    "",
    response_model=PaginatedResponse[UserRead],
)
async def list_users(
    session: SessionDep,
    cursor: str | None = None,
    items_per_page: int = 20,
):
    return await UserCrud.cursor_paginate(
        session=session,
        cursor=cursor,
        items_per_page=items_per_page,
    )

The cursor_paginate method returns a PaginatedResponse whose pagination field is a CursorPagination object:

{
  "status": "SUCCESS",
  "data": ["..."],
  "pagination": {
    "next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
    "prev_cursor": null,
    "items_per_page": 20,
    "has_more": true
  }
}

Pass next_cursor as the cursor query parameter on the next request to advance to the next page. prev_cursor is set on pages 2+ and points back to the first item of the current page. Both are null when there is no adjacent page.

Choosing a cursor column

The cursor column is set once on CrudFactory via the cursor_column parameter. It must be monotonically ordered for stable results:

• Auto-increment integer PKs
• UUID v7 PKs
• Timestamps

!!! warning Random UUID v4 PKs are not suitable as cursor columns because their ordering is non-deterministic.

!!! note cursor_column is required. Calling cursor_paginate on a CRUD class that has no cursor_column configured raises a ValueError.

The cursor value is base64-encoded when returned to the client and decoded back to the correct Python type on the next request. The following SQLAlchemy column types are supported:

SQLAlchemy type	Python type
Integer, BigInteger, SmallInteger	int
Uuid	uuid.UUID
DateTime	datetime.datetime
Date	datetime.date
Float, Numeric	decimal.Decimal

# Paginate by the primary key
PostCrud = CrudFactory(model=Post, cursor_column=Post.id)

# Paginate by a timestamp column instead
PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at)

Search

Two search strategies are available, both compatible with offset_paginate and cursor_paginate.

	Full-text search	Faceted search
Input	Free-text string	Exact column values
Relationship support	Yes	Yes
Use case	Search bars	Filter dropdowns

!!! info "You can use both search strategies in the same endpoint!"

Full-text 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
    ],
)

You can override searchable_fields per call with search_fields:

result = await UserCrud.offset_paginate(
    session=session,
    search_fields=[User.country],
)

This allows searching with both offset_paginate and cursor_paginate:

@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.offset_paginate(
        session=session,
        items_per_page=items_per_page,
        page=page,
        search=search,
    )

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

Faceted search

!!! info "Added in v1.2"

Declare facet_fields on the CRUD class to return distinct column values alongside paginated results. This is useful for populating filter dropdowns or building faceted search UIs.

Facet fields use the same syntax as searchable_fields — direct columns or relationship tuples:

UserCrud = CrudFactory(
    model=User,
    facet_fields=[
        User.status,
        User.country,
        (User.role, Role.name),  # value from a related model
    ],
)

You can override facet_fields per call:

result = await UserCrud.offset_paginate(
    session=session,
    facet_fields=[User.country],
)

The distinct values are returned in the filter_attributes field of PaginatedResponse:

{
  "status": "SUCCESS",
  "data": ["..."],
  "pagination": { "..." },
  "filter_attributes": {
    "status": ["active", "inactive"],
    "country": ["DE", "FR", "US"],
    "name": ["admin", "editor", "viewer"]
  }
}

Use filter_by to pass the client's chosen filter values directly — no need to build SQLAlchemy conditions by hand. Any unknown key raises InvalidFacetFilterError.

!!! info "The keys in filter_by are the same keys the client received in filter_attributes." Keys are normally the terminal column.key (e.g. "name" for Role.name). When two facet fields share the same column key (e.g. (Build.project, Project.name) and (Build.os, Os.name)), the relationship name is prepended automatically: "project__name" and "os__name".

filter_by and filters can be combined — both are applied with AND logic.

Use filter_params() to generate a dict with the facet filter values from the query parameters:

from typing import Annotated

from fastapi import Depends

UserCrud = CrudFactory(
    model=User,
    facet_fields=[User.status, User.country, (User.role, Role.name)],
)

@router.get("", response_model_exclude_none=True)
async def list_users(
    session: SessionDep,
    page: int = 1,
    filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())],
) -> PaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(
        session=session,
        page=page,
        filter_by=filter_by,
    )

Both single-value and multi-value query parameters work:

GET /users?status=active              → filter_by={"status": ["active"]}
GET /users?status=active&country=FR   → filter_by={"status": ["active"], "country": ["FR"]}
GET /users?role=admin&role=editor     → filter_by={"role": ["admin", "editor"]}  (IN clause)

Sorting

!!! info "Added in v1.3"

Declare order_fields on the CRUD class to expose client-driven column ordering via order_by and order query parameters.

UserCrud = CrudFactory(
    model=User,
    order_fields=[
        User.name,
        User.created_at,
    ],
)

Call order_params() to generate a FastAPI dependency that maps the query parameters to an OrderByClause expression:

from typing import Annotated

from fastapi import Depends
from fastapi_toolsets.crud import OrderByClause

@router.get("")
async def list_users(
    session: SessionDep,
    order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())],
) -> PaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, order_by=order_by)

The dependency adds two query parameters to the endpoint:

Parameter	Type
order_by	`str
order	asc or desc

GET /users?order_by=name&order=asc   → ORDER BY users.name ASC
GET /users?order_by=name&order=desc  → ORDER BY users.name DESC

An unknown order_by value raises InvalidOrderFieldError (HTTP 422).

You can also pass order_fields directly to order_params() to override the class-level defaults without modifying them:

UserOrderParams = UserCrud.order_params(order_fields=[User.name])

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, offset_paginate, cursor_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"},
)

Response serialization

!!! info "Added in v1.1"

Pass a Pydantic schema class to create, get, update, or offset_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.offset_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