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) async classmethod 

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

create(
    session: AsyncSession,
    obj: BaseModel,
    *,
    as_response: Literal[False] = ...,
) -> 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

If True, wrap result in Response object

 False

Returns:

Type Description
ModelType | Response[ModelType]

Created model instance or Response wrapping it

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

If True, wrap result in Response object

 False

Returns:

Type Description
bool | Response[None]

True if deletion was executed, or Response wrapping it

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[Any] | 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) async classmethod 

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

get(
    session: AsyncSession,
    filters: list[Any],
    *,
    joins: JoinType | None = None,
    outer_join: bool = False,
    with_for_update: bool = False,
    load_options: list[Any] | None = None,
    as_response: Literal[False] = ...,
) -> 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[Any] | None

SQLAlchemy loader options (e.g., selectinload)

 None
as_response bool

If True, wrap result in Response object

 False

Returns:

Type Description
ModelType | Response[ModelType]

Model instance or Response wrapping it

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[Any] | 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

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) async classmethod

Get paginated results with metadata.

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[Any] | 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

Returns:

Type Description
PaginatedResponse[ModelType]

Dict with 'data' and 'pagination' keys

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

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

update(
    session: AsyncSession,
    obj: BaseModel,
    filters: list[Any],
    *,
    exclude_unset: bool = True,
    exclude_none: bool = False,
    as_response: Literal[False] = ...,
) -> 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

If True, wrap result in Response object

 False

Returns:

Type Description
ModelType | Response[ModelType]

Updated model instance or Response wrapping it

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)

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

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

# 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.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)