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, *, as_response=False, schema=None)` `async` `classmethod`

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

    create(
        session: AsyncSession,
        obj: BaseModel,
        *,
        as_response: Literal[True],
        schema: None = ...,
    ) -> Response[ModelType]

    create(
        session: AsyncSession,
        obj: BaseModel,
        *,
        as_response: Literal[False] = ...,
        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
`as_response` | `bool` | Deprecated. Use `schema` instead. Will be removed in v2.0. | `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[ModelType] | Response[Any]` | Created model instance, or `Response[schema]` when `schema` is given,
`ModelType | Response[ModelType] | Response[Any]` | or `Response[ModelType]` when `as_response=True` (deprecated).

### `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, schema=None)` `async` `classmethod`

    cursor_paginate(
        session: AsyncSession,
        *,
        cursor: str | None = None,
        filters: list[Any] | None = None,
        joins: JoinType | None = None,
        outer_join: bool = False,
        load_options: list[ExecutableOption] | None = None,
        order_by: Any | None = None,
        items_per_page: int = 20,
        search: str | SearchConfig | None = None,
        search_fields: Sequence[SearchFieldType] | None = None,
        schema: type[SchemaType],
    ) -> PaginatedResponse[SchemaType]

    cursor_paginate(
        session: AsyncSession,
        *,
        cursor: str | None = None,
        filters: list[Any] | None = None,
        joins: JoinType | None = None,
        outer_join: bool = False,
        load_options: list[ExecutableOption] | None = None,
        order_by: Any | None = None,
        items_per_page: int = 20,
        search: str | SearchConfig | None = None,
        search_fields: Sequence[SearchFieldType] | None = None,
        schema: None = ...,
    ) -> PaginatedResponse[ModelType]

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` | `list[ExecutableOption] | None` | SQLAlchemy loader options. Falls back to `default_load_options` when not provided. | `None`
`order_by` | `Any | 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`
`schema` | `type[BaseModel] | None` | Optional Pydantic schema to serialize each item into. | `None`

Returns:

Type | Description
--- | ---
`PaginatedResponse[ModelType] | PaginatedResponse[Any]` | PaginatedResponse with CursorPagination metadata

### `delete(session, filters, *, as_response=False)` `async` `classmethod`

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

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

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
`as_response` | `bool` | Deprecated. Will be removed in v2.0. When `True`, returns `Response[None]` instead of `bool`. | `False`

Returns:

Type | Description
--- | ---
`bool | Response[None]` | `True` if deletion was executed, or `Response[None]` when
`bool | Response[None]` | `as_response=True` (deprecated).

### `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, load_options=None)` `async` `classmethod`

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`
`load_options` | `list[ExecutableOption] | None` | SQLAlchemy loader options | `None`

Returns:

Type | Description
--- | ---
`ModelType | None` | Model instance or None

### `get(session, filters, *, joins=None, outer_join=False, with_for_update=False, load_options=None, as_response=False, 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: list[ExecutableOption] | None = None,
        schema: type[SchemaType],
        as_response: bool = ...,
    ) -> Response[SchemaType]

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

    get(
        session: AsyncSession,
        filters: list[Any],
        *,
        joins: JoinType | None = None,
        outer_join: bool = False,
        with_for_update: bool = False,
        load_options: list[ExecutableOption] | None = None,
        as_response: Literal[False] = ...,
        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` | `list[ExecutableOption] | None` | SQLAlchemy loader options (e.g., selectinload) | `None`
`as_response` | `bool` | Deprecated. Use `schema` instead. Will be removed in v2.0. | `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[ModelType] | Response[Any]` | Model instance, or `Response[schema]` when `schema` is given,
`ModelType | Response[ModelType] | Response[Any]` | or `Response[ModelType]` when `as_response=True` (deprecated).

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` | `list[ExecutableOption] | None` | SQLAlchemy loader options | `None`
`order_by` | `Any | 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

### `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, schema=None)` `async` `classmethod`

    offset_paginate(
        session: AsyncSession,
        *,
        filters: list[Any] | None = None,
        joins: JoinType | None = None,
        outer_join: bool = False,
        load_options: list[ExecutableOption] | None = None,
        order_by: Any | None = None,
        page: int = 1,
        items_per_page: int = 20,
        search: str | SearchConfig | None = None,
        search_fields: Sequence[SearchFieldType] | None = None,
        schema: type[SchemaType],
    ) -> PaginatedResponse[SchemaType]

    offset_paginate(
        session: AsyncSession,
        *,
        filters: list[Any] | None = None,
        joins: JoinType | None = None,
        outer_join: bool = False,
        load_options: list[ExecutableOption] | None = None,
        order_by: Any | None = None,
        page: int = 1,
        items_per_page: int = 20,
        search: str | SearchConfig | None = None,
        search_fields: Sequence[SearchFieldType] | None = None,
        schema: None = ...,
    ) -> PaginatedResponse[ModelType]

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` | `list[ExecutableOption] | None` | SQLAlchemy loader options | `None`
`order_by` | `Any | 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`
`schema` | `type[BaseModel] | None` | Optional Pydantic schema to serialize each item into. | `None`

Returns:

Type | Description
--- | ---
`PaginatedResponse[ModelType] | PaginatedResponse[Any]` | PaginatedResponse with OffsetPagination metadata

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

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

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

    update(
        session: AsyncSession,
        obj: BaseModel,
        filters: list[Any],
        *,
        exclude_unset: bool = True,
        exclude_none: bool = False,
        as_response: Literal[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`
`as_response` | `bool` | Deprecated. Use `schema` instead. Will be removed in v2.0. | `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[ModelType] | Response[Any]` | Updated model instance, or `Response[schema]` when `schema` is given,
`ModelType | Response[ModelType] | Response[Any]` | or `Response[ModelType]` when `as_response=True` (deprecated).

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, 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`
`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` | `list[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 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)`

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)