`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, order_joins=None, items_per_page=20, search=None, search_fields=None, search_column=None, order_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`
`search_column`	`str \| None`	Restrict search to a single column key.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Fields allowed for sorting (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
`CursorPaginatedResponse[Any]`	PaginatedResponse with CursorPagination metadata

`cursor_paginate_params(*, default_page_size=20, max_page_size=100, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

Return a FastAPI dependency that collects all params for :meth:cursor_paginate.

Parameters:

Name	Type	Description	Default
`default_page_size`	`int`	Default `items_per_page` value.	`20`
`max_page_size`	`int`	Maximum `items_per_page` value.	`100`
`search`	`bool`	Enable search query parameters.	`True`
`filter`	`bool`	Enable facet filter query parameters.	`True`
`order`	`bool`	Enable order query parameters.	`True`
`search_fields`	`Sequence[SearchFieldType] \| None`	Override searchable fields.	`None`
`facet_fields`	`Sequence[FacetFieldType] \| None`	Override facet fields.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Override order fields.	`None`
`default_order_field`	`QueryableAttribute[Any] \| None`	Default field to order by when `order_by` is absent.	`None`
`default_order`	`Literal['asc', 'desc']`	Default sort direction.	`'asc'`

Returns:

Name	Type	Description
	`Callable[..., Awaitable[dict[str, Any]]]`	An async dependency that resolves to a dict ready to be unpacked
`into`	`Callable[..., Awaitable[dict[str, Any]]]`	meth:`cursor_paginate`.

`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

`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, order_joins=None, page=1, items_per_page=20, include_total=True, search=None, search_fields=None, search_column=None, order_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`
`include_total`	`bool`	When `False`, skip the `COUNT` query; `pagination.total_count` will be `None`.	`True`
`search`	`str \| SearchConfig \| None`	Search query string or SearchConfig object	`None`
`search_fields`	`Sequence[SearchFieldType] \| None`	Fields to search in (overrides class default)	`None`
`search_column`	`str \| None`	Restrict search to a single column key.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Fields allowed for sorting (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
`OffsetPaginatedResponse[Any]`	PaginatedResponse with OffsetPagination metadata

`offset_paginate_params(*, default_page_size=20, max_page_size=100, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

Return a FastAPI dependency that collects all params for :meth:offset_paginate.

Parameters:

Name	Type	Description	Default
`default_page_size`	`int`	Default `items_per_page` value.	`20`
`max_page_size`	`int`	Maximum `items_per_page` value.	`100`
`include_total`	`bool`	Whether to include total count (not a query param).	`True`
`search`	`bool`	Enable search query parameters.	`True`
`filter`	`bool`	Enable facet filter query parameters.	`True`
`order`	`bool`	Enable order query parameters.	`True`
`search_fields`	`Sequence[SearchFieldType] \| None`	Override searchable fields.	`None`
`facet_fields`	`Sequence[FacetFieldType] \| None`	Override facet fields.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Override order fields.	`None`
`default_order_field`	`QueryableAttribute[Any] \| None`	Default field to order by when `order_by` is absent.	`None`
`default_order`	`Literal['asc', 'desc']`	Default sort direction.	`'asc'`

Returns:

Name	Type	Description
	`Callable[..., Awaitable[dict[str, Any]]]`	An async dependency that resolves to a dict ready to be unpacked
`into`	`Callable[..., Awaitable[dict[str, Any]]]`	meth:`offset_paginate`.

`paginate(session, *, pagination_type=PaginationType.OFFSET, filters=None, joins=None, outer_join=False, load_options=None, order_by=None, order_joins=None, page=1, cursor=None, items_per_page=20, include_total=True, search=None, search_fields=None, search_column=None, order_fields=None, facet_fields=None, filter_by=None, schema)` `async` `classmethod` ¶

paginate(
    session: AsyncSession,
    *,
    pagination_type: Literal[PaginationType.OFFSET],
    filters: list[Any] | None = ...,
    joins: JoinType | None = ...,
    outer_join: bool = ...,
    load_options: Sequence[ExecutableOption] | None = ...,
    order_by: OrderByClause | None = ...,
    order_joins: list[Any] | None = ...,
    page: int = ...,
    cursor: str | None = ...,
    items_per_page: int = ...,
    include_total: bool = ...,
    search: str | SearchConfig | None = ...,
    search_fields: Sequence[SearchFieldType] | None = ...,
    search_column: str | None = ...,
    order_fields: Sequence[OrderFieldType] | None = ...,
    facet_fields: Sequence[FacetFieldType] | None = ...,
    filter_by: dict[str, Any] | BaseModel | None = ...,
    schema: type[BaseModel],
) -> OffsetPaginatedResponse[Any]

paginate(
    session: AsyncSession,
    *,
    pagination_type: Literal[PaginationType.CURSOR],
    filters: list[Any] | None = ...,
    joins: JoinType | None = ...,
    outer_join: bool = ...,
    load_options: Sequence[ExecutableOption] | None = ...,
    order_by: OrderByClause | None = ...,
    order_joins: list[Any] | None = ...,
    page: int = ...,
    cursor: str | None = ...,
    items_per_page: int = ...,
    include_total: bool = ...,
    search: str | SearchConfig | None = ...,
    search_fields: Sequence[SearchFieldType] | None = ...,
    search_column: str | None = ...,
    order_fields: Sequence[OrderFieldType] | None = ...,
    facet_fields: Sequence[FacetFieldType] | None = ...,
    filter_by: dict[str, Any] | BaseModel | None = ...,
    schema: type[BaseModel],
) -> CursorPaginatedResponse[Any]

Get paginated results using either offset or cursor pagination.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	DB async session.	required
`pagination_type`	`PaginationType`	Pagination strategy. Defaults to `PaginationType.OFFSET`.	`OFFSET`
`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`	Column or expression to order results by.	`None`
`page`	`int`	Page number (1-indexed). Only used when `pagination_type` is `OFFSET`.	`1`
`cursor`	`str \| None`	Cursor token from a previous :class:`.CursorPaginatedResponse`. Only used when `pagination_type` is `CURSOR`.	`None`
`items_per_page`	`int`	Number of items per page (default 20).	`20`
`include_total`	`bool`	When `False`, skip the `COUNT` query; only applies when `pagination_type` is `OFFSET`.	`True`
`search`	`str \| SearchConfig \| None`	Search query string or :class:`.SearchConfig` object.	`None`
`search_fields`	`Sequence[SearchFieldType] \| None`	Fields to search in (overrides class default).	`None`
`search_column`	`str \| None`	Restrict search to a single column key.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Fields allowed for sorting (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 :exc:`.InvalidFacetFilterError` for unknown keys.	`None`
`schema`	`type[BaseModel]`	Pydantic schema to serialize each item into.	required

Returns:

Type	Description
`OffsetPaginatedResponse[Any] \| CursorPaginatedResponse[Any]`	class:`.OffsetPaginatedResponse` when `pagination_type` is
`OffsetPaginatedResponse[Any] \| CursorPaginatedResponse[Any]`	`OFFSET`, :class:`.CursorPaginatedResponse` when it is
`OffsetPaginatedResponse[Any] \| CursorPaginatedResponse[Any]`	`CURSOR`.

`paginate_params(*, default_page_size=20, max_page_size=100, default_pagination_type=PaginationType.OFFSET, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

Return a FastAPI dependency that collects all params for :meth:paginate.

Parameters:

Name	Type	Description	Default
`default_page_size`	`int`	Default `items_per_page` value.	`20`
`max_page_size`	`int`	Maximum `items_per_page` value.	`100`
`default_pagination_type`	`PaginationType`	Default pagination strategy.	`OFFSET`
`include_total`	`bool`	Whether to include total count (not a query param).	`True`
`search`	`bool`	Enable search query parameters.	`True`
`filter`	`bool`	Enable facet filter query parameters.	`True`
`order`	`bool`	Enable order query parameters.	`True`
`search_fields`	`Sequence[SearchFieldType] \| None`	Override searchable fields.	`None`
`facet_fields`	`Sequence[FacetFieldType] \| None`	Override facet fields.	`None`
`order_fields`	`Sequence[OrderFieldType] \| None`	Override order fields.	`None`
`default_order_field`	`QueryableAttribute[Any] \| None`	Default field to order by when `order_by` is absent.	`None`
`default_order`	`Literal['asc', 'desc']`	Default sort direction.	`'asc'`

Returns:

Name	Type	Description
	`Callable[..., Awaitable[dict[str, Any]]]`	An async dependency that resolves to a dict ready to be unpacked
`into`	`Callable[..., Awaitable[dict[str, Any]]]`	meth:`paginate`.

`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, *, base_class=AsyncCrud, 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
`base_class`	`type[AsyncCrud[Any]]`	Optional base class to inherit from instead of `AsyncCrud`. Use this to share custom methods across multiple CRUD classes while still using the factory shorthand.	`AsyncCrud`
`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[OrderFieldType] \| None`	Optional list of model attributes that callers are allowed to order by via `offset_paginate_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,
)

# With a shared custom base class:
from typing import Generic, TypeVar
from sqlalchemy.orm import DeclarativeBase

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

class AuditedCrud(AsyncCrud[T], Generic[T]):
    @classmethod
    async def get_active(cls, session):
        return await cls.get_multi(session, filters=[cls.model.is_active == True])

UserCrud = CrudFactory(User, base_class=AuditedCrud)

`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, search_column=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`
`search_column`	`str \| None`	Optional key to narrow search to a single field. Must match one of the resolved search field keys.	`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

crud¶

fastapi_toolsets.crud.factory.AsyncCrud ¶

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

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

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

cursor_paginate_params(*, default_page_size=20, max_page_size=100, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc') classmethod ¶

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

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

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

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

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

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

offset_paginate_params(*, default_page_size=20, max_page_size=100, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc') classmethod ¶

paginate_params(*, default_page_size=20, max_page_size=100, default_pagination_type=PaginationType.OFFSET, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc') classmethod ¶

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

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

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

fastapi_toolsets.crud.search.SearchConfig dataclass ¶

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

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

`crud`¶

`fastapi_toolsets.crud.factory.AsyncCrud` ¶

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

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

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

`cursor_paginate_params(*, default_page_size=20, max_page_size=100, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

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

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

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

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

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

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

`offset_paginate_params(*, default_page_size=20, max_page_size=100, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

`paginate_params(*, default_page_size=20, max_page_size=100, default_pagination_type=PaginationType.OFFSET, include_total=True, search=True, filter=True, order=True, search_fields=None, facet_fields=None, order_fields=None, default_order_field=None, default_order='asc')` `classmethod` ¶

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

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

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

`fastapi_toolsets.crud.search.SearchConfig` `dataclass` ¶

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

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