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, a 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¶

Factory style¶

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. This is the most concise option for straightforward CRUD with no custom logic.

Subclass style¶

Added in v2.3.0

from fastapi_toolsets.crud.factory import AsyncCrud
from myapp.models import User

class UserCrud(AsyncCrud[User]):
    model = User
    searchable_fields = [User.username, User.email]
    default_load_options = [selectinload(User.role)]

Subclassing AsyncCrud directly is the preferred style when you need to add custom methods or when the configuration is complex enough to benefit from a named class body.

Adding custom methods¶

class UserCrud(AsyncCrud[User]):
    model = User

    @classmethod
    async def get_active(cls, session: AsyncSession) -> list[User]:
        return await cls.get_multi(session, filters=[User.is_active == True])

Define a generic base class with the shared methods, then subclass it for each model:

from typing import Generic, TypeVar
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase
from fastapi_toolsets.crud.factory import AsyncCrud

T = TypeVar("T", bound=DeclarativeBase)

class AuditedCrud(AsyncCrud[T], Generic[T]):
    """Base CRUD with custom function"""

    @classmethod
    async def get_active(cls, session: AsyncSession):
        return await cls.get_multi(session, filters=[cls.model.is_active == True])


class UserCrud(AuditedCrud[User]):
    model = User
    searchable_fields = [User.username, User.email]

You can also use the factory shorthand with the same base by passing base_class:

UserCrud = CrudFactory(User, base_class=AuditedCrud)

Basic operations¶

get_or_none added in v2.2

# 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 one or None (never raises)
user = await UserCrud.get_or_none(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])

Fetching a single record¶

Three methods fetch a single record — choose based on how you want to handle the "not found" case and whether you need strict uniqueness:

Method	Not found	Multiple results
`get`	raises `NotFoundError`	raises `MultipleResultsFound`
`get_or_none`	returns `None`	raises `MultipleResultsFound`
`first`	returns `None`	returns the first match silently

Use get when the record must exist (e.g. a detail endpoint that should return 404):

user = await UserCrud.get(session=session, filters=[User.id == user_id])

Use get_or_none when the record may not exist but you still want strict uniqueness enforcement:

user = await UserCrud.get_or_none(session=session, filters=[User.email == email])
if user is None:
    ...  # handle missing case without catching an exception

Use first when you only care about any one match and don't need uniqueness:

user = await UserCrud.first(session=session, filters=[User.is_active == True])

Pagination¶

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

Three pagination methods are available. All return a typed response whose pagination_type field tells clients which strategy was used.

	`offset_paginate`	`cursor_paginate`	`paginate`
Return type	`OffsetPaginatedResponse`	`CursorPaginatedResponse`	either, based on `pagination_type` param
Total count	Yes	No	/
Jump to arbitrary page	Yes	No	/
Performance on deep pages	Degrades	Constant	/
Stable under concurrent inserts	No	Yes	/
Use case	Admin panels, numbered pagination	Feeds, APIs, infinite scroll	single endpoint, both strategies

Offset pagination¶

from typing import Annotated
from fastapi import Depends

@router.get("")
async def get_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)

The offset_paginate method returns an OffsetPaginatedResponse:

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

Skipping the COUNT query¶

Added in v2.4.1

By default offset_paginate runs two queries: one for the page items and one COUNT(*) for total_count. On large tables the COUNT can be expensive. Pass include_total=False to offset_paginate_params() to skip it:

@router.get("")
async def get_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.offset_paginate_params(include_total=False))],
) -> OffsetPaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)

Cursor pagination¶

@router.get("")
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
    return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)

The cursor_paginate method returns a CursorPaginatedResponse:

{
  "status": "SUCCESS",
  "pagination_type": "cursor",
  "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 URL-safe base64-encoded (no padding) 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)

Unified endpoint (both strategies)¶

Added in v2.3.0

paginate() dispatches to offset_paginate or cursor_paginate based on a pagination_type query parameter, letting you expose one endpoint that supports both strategies. The pagination_type field in the response tells clients which strategy was used, enabling frontend discriminated-union typing.

from fastapi_toolsets.schemas import PaginatedResponse

@router.get("")
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.paginate_params())],
) -> PaginatedResponse[UserRead]:
    return await UserCrud.paginate(session, **params, schema=UserRead)

GET /users?pagination_type=offset&page=2&items_per_page=10
GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10

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

You can use both search strategies in the same endpoint!

Full-text search¶

Added in v2.2.1

The model's primary key is always included in searchable_fields automatically, so searching by ID works out of the box without any configuration. When no searchable_fields are declared, only the primary key is searched.

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

Or via the dependency to narrow which fields are exposed as query parameters:

params = UserCrud.offset_paginate_params(search_fields=[Post.title])

This allows searching with both offset_paginate and cursor_paginate:

@router.get("")
async def get_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)

@router.get("")
async def get_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
    return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)

The dependency adds two query parameters to the endpoint:

Parameter	Type
`search`	`str \\| null`
`search_column`	`str \\| null`

GET /posts?search=hello                        → search all configured columns
GET /posts?search=hello&search_column=title    → search only Post.title

The available search column keys are returned in the search_columns field of PaginatedResponse. Use them to populate a column picker in the UI, or to validate search_column values on the client side:

{
  "status": "SUCCESS",
  "data": ["..."],
  "pagination": { "..." },
  "search_columns": ["content", "author__username", "title"]
}

Key format uses __ as a separator for relationship chains.

A direct column Post.title produces "title". A relationship tuple (Post.author, User.username) produces "author__username". An unknown search_column value raises InvalidSearchColumnError (HTTP 422).

Faceted search¶

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. Relationship traversal is supported via tuples, using the same syntax as searchable_fields:

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

Or via the dependency to narrow which fields are exposed as query parameters:

params = UserCrud.offset_paginate_params(facet_fields=[User.country])

Facet filtering is built into the consolidated params dependencies. When filter=True (the default), each facet field is exposed as a query parameter and values are collected into filter_by automatically:

from typing import Annotated

from fastapi import Depends

@router.get("", response_model_exclude_none=True)
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)

@router.get("", response_model_exclude_none=True)
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
    return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)

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__name=admin&role__name=editor → filter_by={"role__name": ["admin", "editor"]}  (IN clause)

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

The distinct values for each facet field are returned in the filter_attributes field of PaginatedResponse. Use them to populate filter dropdowns in the UI, or to validate filter_by keys on the client side:

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

Key format uses __ as a separator for relationship chains.

A direct column User.status produces "status". A relationship tuple (User.role, Role.name) produces "role__name". A deeper chain (User.role, Role.permission, Permission.name) produces "role__permission__name". An unknown filter_by key raises InvalidFacetFilterError (HTTP 422).

Sorting¶

Added in v1.3

Declare order_fields on the CRUD class. Relationship traversal is supported via tuples, using the same syntax as searchable_fields and facet_fields:

UserCrud = CrudFactory(
    model=User,
    order_fields=[
        User.name,
        User.created_at,
        (User.role, Role.name),  # sort by a related model column
    ],
)

You can override order_fields per call:

result = await UserCrud.offset_paginate(
    session=session,
    order_fields=[User.name],
)

Or via the dependency to narrow which fields are exposed as query parameters:

params = UserCrud.offset_paginate_params(order_fields=[User.name])

Sorting is built into the consolidated params dependencies. When order=True (the default), order_by and order query parameters are exposed and resolved into an OrderByClause automatically:

from typing import Annotated

from fastapi import Depends

@router.get("")
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
    return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)

@router.get("")
async def list_users(
    session: SessionDep,
    params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
    return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)

The dependency adds two query parameters to the endpoint:

Parameter	Type
`order_by`	`str \\| null`
`order`	`asc` or `desc`

GET /users?order_by=name&order=asc         → ORDER BY users.name ASC
GET /users?order_by=role__name&order=desc  → LEFT JOIN roles ON ... ORDER BY roles.name DESC

Relationship tuples are joined automatically.

When a relation field is selected, the related table is LEFT OUTER JOINed automatically. An unknown order_by value raises InvalidOrderFieldError (HTTP 422).

The available sort keys are returned in the order_columns field of PaginatedResponse. Use them to populate a sort picker in the UI, or to validate order_by values on the client side:

{
  "status": "SUCCESS",
  "data": ["..."],
  "pagination": { "..." },
  "order_columns": ["created_at", "name", "role__name"]
}

Key format uses __ as a separator for relationship chains.

A direct column User.name produces "name". A relationship tuple (User.role, Role.name) produces "role__name".

Relationship loading¶

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¶

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,
    params: Annotated[dict, Depends(crud.UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
    return await crud.UserCrud.offset_paginate(session=session, **params, schema=UserRead)

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

API Reference

CRUD¶

Overview¶

Creating a CRUD class¶

Factory style¶

Subclass style¶

Adding custom methods¶

Sharing a custom base across multiple models¶

Basic operations¶

Fetching a single record¶

Pagination¶

Offset pagination¶

Skipping the COUNT query¶

Cursor pagination¶

Choosing a cursor column¶

Unified endpoint (both strategies)¶

Search¶

Full-text search¶

Faceted search¶

Sorting¶

Relationship loading¶

Many-to-many relationships¶

Upsert¶

Response serialization¶