Skip to content

crud

Here's the reference for the CRUD classes, factory, and search utilities.

You can import the main symbols from fastapi_toolsets.crud:

from fastapi_toolsets.crud import CrudFactory, AsyncCrud
from fastapi_toolsets.crud.search import SearchConfig, get_searchable_fields, build_search_filters

fastapi_toolsets.crud.factory.AsyncCrud

Bases: Generic[ModelType]

Generic async CRUD operations for SQLAlchemy models.

Subclass this and set the model class variable, or use CrudFactory.

count(session, filters=None, *, joins=None, outer_join=False) async classmethod

Count records matching the filters.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any] | None

List of SQLAlchemy filter conditions

 None
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False

Returns:

Type Description
int

Number of matching records

create(session, obj, *, schema=None) async classmethod 

create(
    session: AsyncSession,
    obj: BaseModel,
    *,
    schema: type[SchemaType],
) -> Response[SchemaType]

create(
    session: AsyncSession,
    obj: BaseModel,
    *,
    schema: None = ...,
) -> ModelType

Create a new record in the database.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
obj BaseModel

Pydantic model with data to create

 required
schema type[BaseModel] | None

Pydantic schema to serialize the result into. When provided, the result is automatically wrapped in a Response[schema].

 None

Returns:

Type Description
ModelType | Response[Any]

Created model instance, or Response[schema] when schema is given.

cursor_paginate(session, *, cursor=None, filters=None, joins=None, outer_join=False, load_options=None, order_by=None, items_per_page=20, search=None, search_fields=None, facet_fields=None, filter_by=None, schema) async classmethod

Get paginated results using cursor-based pagination.

Parameters:

Name Type Description Default
session AsyncSession

DB async session.

 required
cursor str | None

Cursor string from a previous CursorPagination. Omit (or pass None) to start from the beginning.

 None
filters list[Any] | None

List of SQLAlchemy filter conditions.

 None
joins JoinType | None

List of (model, condition) tuples for joining related tables.

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN.

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options. Falls back to default_load_options when not provided.

 None
order_by OrderByClause | None

Additional ordering applied after the cursor column.

 None
items_per_page int

Number of items per page (default 20).

 20
search str | SearchConfig | None

Search query string or SearchConfig object.

 None
search_fields Sequence[SearchFieldType] | None

Fields to search in (overrides class default).

 None
facet_fields Sequence[FacetFieldType] | None

Columns to compute distinct values for (overrides class default).

 None
filter_by dict[str, Any] | BaseModel | None

Dict of {column_key: value} to filter by declared facet fields. Keys must match the column.key of a facet field. Scalar → equality, list → IN clause. Raises InvalidFacetFilterError for unknown keys.

 None
schema type[BaseModel]

Optional Pydantic schema to serialize each item into.

 required

Returns:

Type Description
PaginatedResponse[Any]

PaginatedResponse with CursorPagination metadata

delete(session, filters, *, return_response=False) async classmethod 

delete(
    session: AsyncSession,
    filters: list[Any],
    *,
    return_response: Literal[True],
) -> Response[None]

delete(
    session: AsyncSession,
    filters: list[Any],
    *,
    return_response: Literal[False] = ...,
) -> None

Delete records from the database.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any]

List of SQLAlchemy filter conditions

 required
return_response bool

When True, returns Response[None] instead of None. Useful for API endpoints that expect a consistent response envelope.

 False

Returns:

Type Description
None | Response[None]

None, or Response[None] when return_response=True.

exists(session, filters, *, joins=None, outer_join=False) async classmethod

Check if a record exists.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any]

List of SQLAlchemy filter conditions

 required
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False

Returns:

Type Description
bool

True if at least one record matches

filter_params(*, facet_fields=None) classmethod

Return a FastAPI dependency that collects facet filter values from query parameters. Args: facet_fields: Override the facet fields for this dependency. Falls back to the class-level facet_fields if not provided.

Returns:

Type Description
Callable[..., Awaitable[dict[str, list[str]]]]

An async dependency function named {Model}FilterParams that resolves to a
Callable[..., Awaitable[dict[str, list[str]]]]

dict[str, list[str]] containing only the keys that were supplied in the
Callable[..., Awaitable[dict[str, list[str]]]]

request (absent/None parameters are excluded).

Raises:

Type Description
ValueError

If no facet fields are configured on this CRUD class and none are provided via facet_fields.

first(session, filters=None, *, joins=None, outer_join=False, with_for_update=False, load_options=None, schema=None) async classmethod 

first(
    session: AsyncSession,
    filters: list[Any] | None = None,
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: type[SchemaType],
) -> Response[SchemaType] | None

first(
    session: AsyncSession,
    filters: list[Any] | None = None,
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: None = ...,
) -> ModelType | None

Get the first matching record, or None.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any] | None

List of SQLAlchemy filter conditions

 None
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False
with_for_update bool

Lock the row for update

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options (e.g., selectinload)

 None
schema type[BaseModel] | None

Pydantic schema to serialize the result into. When provided, the result is automatically wrapped in a Response[schema].

 None

Returns:

Type Description
ModelType | Response[Any] | None

Model instance, Response[schema] when schema is given,
ModelType | Response[Any] | None

or None when no record matches.

get(session, filters, *, joins=None, outer_join=False, with_for_update=False, load_options=None, schema=None) async classmethod 

get(
    session: AsyncSession,
    filters: list[Any],
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: type[SchemaType],
) -> Response[SchemaType]

get(
    session: AsyncSession,
    filters: list[Any],
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: None = ...,
) -> ModelType

Get exactly one record. Raises NotFoundError if not found.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any]

List of SQLAlchemy filter conditions

 required
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False
with_for_update bool

Lock the row for update

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options (e.g., selectinload)

 None
schema type[BaseModel] | None

Pydantic schema to serialize the result into. When provided, the result is automatically wrapped in a Response[schema].

 None

Returns:

Type Description
ModelType | Response[Any]

Model instance, or Response[schema] when schema is given.

Raises:

Type Description
NotFoundError

If no record found
MultipleResultsFound

If more than one record found

get_multi(session, *, filters=None, joins=None, outer_join=False, load_options=None, order_by=None, limit=None, offset=None) async classmethod

Get multiple records from the database.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any] | None

List of SQLAlchemy filter conditions

 None
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options

 None
order_by OrderByClause | None

Column or list of columns to order by

 None
limit int | None

Max number of rows to return

 None
offset int | None

Rows to skip

 None

Returns:

Type Description
Sequence[ModelType]

List of model instances

get_or_none(session, filters, *, joins=None, outer_join=False, with_for_update=False, load_options=None, schema=None) async classmethod 

get_or_none(
    session: AsyncSession,
    filters: list[Any],
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: type[SchemaType],
) -> Response[SchemaType] | None

get_or_none(
    session: AsyncSession,
    filters: list[Any],
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: Sequence[ExecutableOption] | None = None,
    schema: None = ...,
) -> ModelType | None

Get exactly one record, or None if not found.

Like :meth:get but returns None instead of raising :class:~fastapi_toolsets.exceptions.NotFoundError when no record matches the filters.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any]

List of SQLAlchemy filter conditions

 required
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False
with_for_update bool

Lock the row for update

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options (e.g., selectinload)

 None
schema type[BaseModel] | None

Pydantic schema to serialize the result into. When provided, the result is automatically wrapped in a Response[schema].

 None

Returns:

Type Description
ModelType | Response[Any] | None

Model instance, Response[schema] when schema is given,
ModelType | Response[Any] | None

or None when no record matches.

Raises:

Type Description
MultipleResultsFound

If more than one record found

offset_paginate(session, *, filters=None, joins=None, outer_join=False, load_options=None, order_by=None, page=1, items_per_page=20, search=None, search_fields=None, facet_fields=None, filter_by=None, schema) async classmethod

Get paginated results using offset-based pagination.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
filters list[Any] | None

List of SQLAlchemy filter conditions

 None
joins JoinType | None

List of (model, condition) tuples for joining related tables

 None
outer_join bool

Use LEFT OUTER JOIN instead of INNER JOIN

 False
load_options Sequence[ExecutableOption] | None

SQLAlchemy loader options

 None
order_by OrderByClause | None

Column or list of columns to order by

 None
page int

Page number (1-indexed)

 1
items_per_page int

Number of items per page

 20
search str | SearchConfig | None

Search query string or SearchConfig object

 None
search_fields Sequence[SearchFieldType] | None

Fields to search in (overrides class default)

 None
facet_fields Sequence[FacetFieldType] | None

Columns to compute distinct values for (overrides class default)

 None
filter_by dict[str, Any] | BaseModel | None

Dict of {column_key: value} to filter by declared facet fields. Keys must match the column.key of a facet field. Scalar → equality, list → IN clause. Raises InvalidFacetFilterError for unknown keys.

 None
schema type[BaseModel]

Pydantic schema to serialize each item into.

 required

Returns:

Type Description
PaginatedResponse[Any]

PaginatedResponse with OffsetPagination metadata

order_params(*, order_fields=None, default_field=None, default_order='asc') classmethod

Return a FastAPI dependency that resolves order query params into an order_by clause.

Parameters:

Name Type Description Default
order_fields Sequence[QueryableAttribute[Any]] | None

Override the allowed order fields. Falls back to the class-level order_fields if not provided.

 None
default_field QueryableAttribute[Any] | None

Field to order by when order_by query param is absent. If None and no order_by is provided, no ordering is applied.

 None
default_order Literal['asc', 'desc']

Default order direction when order is absent ("asc" or "desc").

 'asc'

Returns:

Type Description
Callable[..., Awaitable[OrderByClause | None]]

An async dependency function named {Model}OrderParams that resolves to an
Callable[..., Awaitable[OrderByClause | None]]

OrderByClause (or None). Pass it to Depends() in your route.

Raises:

Type Description
ValueError

If no order fields are configured on this CRUD class and none are provided via order_fields.
InvalidOrderFieldError

When the request provides an unknown order_by value.

update(session, obj, filters, *, exclude_unset=True, exclude_none=False, schema=None) async classmethod 

update(
    session: AsyncSession,
    obj: BaseModel,
    filters: list[Any],
    *,
    exclude_unset: bool = True,
    exclude_none: bool = False,
    schema: type[SchemaType],
) -> Response[SchemaType]

update(
    session: AsyncSession,
    obj: BaseModel,
    filters: list[Any],
    *,
    exclude_unset: bool = True,
    exclude_none: bool = False,
    schema: None = ...,
) -> ModelType

Update a record in the database.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
obj BaseModel

Pydantic model with update data

 required
filters list[Any]

List of SQLAlchemy filter conditions

 required
exclude_unset bool

Exclude fields not explicitly set in the schema

 True
exclude_none bool

Exclude fields with None value

 False
schema type[BaseModel] | None

Pydantic schema to serialize the result into. When provided, the result is automatically wrapped in a Response[schema].

 None

Returns:

Type Description
ModelType | Response[Any]

Updated model instance, or Response[schema] when schema is given.

Raises:

Type Description
NotFoundError

If no record found

upsert(session, obj, index_elements, *, set_=None, where=None) async classmethod

Create or update a record (PostgreSQL only).

Uses INSERT ... ON CONFLICT for atomic upsert.

Parameters:

Name Type Description Default
session AsyncSession

DB async session

 required
obj BaseModel

Pydantic model with data

 required
index_elements list[str]

Columns for ON CONFLICT (unique constraint)

 required
set_ BaseModel | None

Pydantic model for ON CONFLICT DO UPDATE SET

 None
where WhereHavingRole | None

WHERE clause for ON CONFLICT DO UPDATE

 None

Returns:

Type Description
ModelType | None

Model instance

fastapi_toolsets.crud.factory.CrudFactory(model, *, searchable_fields=None, facet_fields=None, order_fields=None, m2m_fields=None, default_load_options=None, cursor_column=None)

Create a CRUD class for a specific model.

Parameters:

Name Type Description Default
model type[ModelType]

SQLAlchemy model class

 required
searchable_fields Sequence[SearchFieldType] | None

Optional list of searchable fields

 None
facet_fields Sequence[FacetFieldType] | None

Optional list of columns to compute distinct values for in paginated responses. Supports direct columns (User.status) and relationship tuples ((User.role, Role.name)). Can be overridden per call.

 None
order_fields Sequence[QueryableAttribute[Any]] | None

Optional list of model attributes that callers are allowed to order by via order_params(). Can be overridden per call.

 None
m2m_fields M2MFieldType | None

Optional mapping for many-to-many relationships. Maps schema field names (containing lists of IDs) to SQLAlchemy relationship attributes.

 None
default_load_options Sequence[ExecutableOption] | None

Default SQLAlchemy loader options applied to all read queries when no explicit load_options are passed. Use this instead of lazy="selectin" on the model so that loading strategy is explicit and per-CRUD. Overridden entirely (not merged) when load_options is provided at call-site.

 None
cursor_column Any | None

Required to call cursor_paginate. Must be monotonically ordered (e.g. integer PK, UUID v7, timestamp). See the cursor pagination docs for supported column types.

 None

Returns:

Type Description
type[AsyncCrud[ModelType]]

AsyncCrud subclass bound to the model
Example 
from fastapi_toolsets.crud import CrudFactory
from myapp.models import User, Post

UserCrud = CrudFactory(User)
PostCrud = CrudFactory(Post)

# With searchable fields:
UserCrud = CrudFactory(
    User,
    searchable_fields=[User.username, User.email, (User.role, Role.name)]
)

# With many-to-many fields:
# Schema has `tag_ids: list[UUID]`, model has `tags` relationship to Tag
PostCrud = CrudFactory(
    Post,
    m2m_fields={"tag_ids": Post.tags},
)

# With facet fields for filter dropdowns / faceted search:
UserCrud = CrudFactory(
    User,
    facet_fields=[User.status, User.country, (User.role, Role.name)],
)

# With a fixed cursor column for cursor_paginate:
PostCrud = CrudFactory(
    Post,
    cursor_column=Post.created_at,
)

# With default load strategy (replaces lazy="selectin" on the model):
ArticleCrud = CrudFactory(
    Article,
    default_load_options=[selectinload(Article.category), selectinload(Article.tags)],
)

# Override default_load_options for a specific call:
article = await ArticleCrud.get(
    session,
    [Article.id == 1],
    load_options=[selectinload(Article.category)],  # tags won't load
)

# Usage
user = await UserCrud.get(session, [User.id == 1])
posts = await PostCrud.get_multi(session, filters=[Post.user_id == user.id])

# Create with M2M - tag_ids are automatically resolved
post = await PostCrud.create(session, PostCreate(title="Hello", tag_ids=[id1, id2]))

# With search
result = await UserCrud.offset_paginate(session, search="john")

# With joins (inner join by default):
users = await UserCrud.get_multi(
    session,
    joins=[(Post, Post.user_id == User.id)],
    filters=[Post.published == True],
)

# With outer join:
users = await UserCrud.get_multi(
    session,
    joins=[(Post, Post.user_id == User.id)],
    outer_join=True,
)

fastapi_toolsets.crud.search.SearchConfig dataclass

Advanced search configuration.

Attributes:

Name Type Description
query str

The search string
fields Sequence[SearchFieldType] | None

Fields to search (columns or tuples for relationships)
case_sensitive bool

Case-sensitive search (default: False)
match_mode Literal['any', 'all']

"any" (OR) or "all" (AND) to combine fields

fastapi_toolsets.crud.search.get_searchable_fields(model, *, include_relationships=True, max_depth=1) cached

Auto-detect String fields on a model and its relationships.

Parameters:

Name Type Description Default
model type[DeclarativeBase]

SQLAlchemy model class

 required
include_relationships bool

Include fields from many-to-one/one-to-one relationships

 True
max_depth int

Max depth for relationship traversal (default: 1)

 1

Returns:

Type Description
list[SearchFieldType]

List of columns and tuples (relationship, column)

fastapi_toolsets.crud.search.build_search_filters(model, search, search_fields=None, default_fields=None)

Build SQLAlchemy filter conditions for search.

Parameters:

Name Type Description Default
model type[DeclarativeBase]

SQLAlchemy model class

 required
search str | SearchConfig

Search string or SearchConfig

 required
search_fields Sequence[SearchFieldType] | None

Fields specified per-call (takes priority)

 None
default_fields Sequence[SearchFieldType] | None

Default fields (from ClassVar)

 None

Returns:

Type Description
tuple[list[ColumnElement[bool]], list[InstrumentedAttribute[Any]]]

Tuple of (filter_conditions, joins_needed)

Raises:

Type Description
NoSearchableFieldsError

If no searchable field has been configured