Skip to content

# CRUD

Generic async CRUD operations for SQLAlchemy models with search, pagination, and many-to-many support. This module has features that are only compatible with Postgres.

Info

This module has been coded and tested to be compatible with PostgreSQL only.

## Overview

The `crud` module provides `AsyncCrud`, an abstract 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

    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.

## Basic operations

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

## Pagination

    @router.get(
        "",
        response_model=PaginatedResponse[User],
    )
    async def get_users(
        session: SessionDep,
        items_per_page: int = 50,
        page: int = 1,
    ):
        return await crud.UserCrud.paginate(
            session=session,
            items_per_page=items_per_page,
            page=page,
        )

The `paginate` function will return a `PaginatedResponse`.

## Search

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

This allow to do a search with the `paginate` function:

    @router.get(
        "",
        response_model=PaginatedResponse[User],
    )
    async def get_users(
        session: SessionDep,
        items_per_page: int = 50,
        page: int = 1,
        search: str | None = None,
    ):
        return await crud.UserCrud.paginate(
            session=session,
            items_per_page=items_per_page,
            page=page,
            search=search,
        )

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

## `as_response`

Pass `as_response=True` to any write operation to get a `Response[ModelType]` back directly for API usage:

    @router.get(
        "/{uuid}",
        response_model=Response[User],
        responses=generate_error_responses(NotFoundError),
    )
    async def get_user(session: SessionDep, uuid: UUID):
        return await crud.UserCrud.get(
            session=session,
            filters=[User.id == uuid],
            as_response=True,
        )

API Reference