How to Extend FastCRUD Functionalities in FastAPI Boilerplate

Extend FastCRUD by subclassing the generic CRUD class to inject custom business logic while preserving all type-safe database operations.

The benavlabs/fastapi-boilerplate leverages FastCRUD as its core database abstraction layer, providing generic, type-safe wrappers around SQLAlchemy async sessions. When you need to extend FastCRUD functionalities beyond the standard get, create, update, and delete operations, you can subclass the generated CRUD classes without modifying the core boilerplate architecture.

Understanding FastCRUD Architecture

In src/app/crud/crud_users.py, the boilerplate instantiates FastCRUD using generic type parameters that bind the SQLAlchemy model to its corresponding Pydantic schemas:

from fastcrud import FastCRUD
from app.models.user import User
from app.schemas.user import (
    UserCreateInternal, UserUpdate, UserUpdateInternal,
    UserDelete, UserRead,
)

CRUDUser = FastCRUD[
    User,
    UserCreateInternal,
    UserUpdate,
    UserUpdateInternal,
    UserDelete,
    UserRead,
]
crud_users = CRUDUser(User)

This generic definition establishes type safety across all database operations. The FastCRUD class provides methods like get, get_multi, create, update, delete, db_delete, exists, and count, along with pagination utilities and joined query helpers documented in docs/user-guide/database/crud.md.

Extend FastCRUD with Custom Methods

To extend FastCRUD functionalities, create a subclass of the generated CRUD class and define async methods that accept db: AsyncSession alongside your custom parameters.

Implementing Soft-Restore Logic

Create src/app/crud/custom_user.py to add a restore method for soft-deleted records:

from fastcrud import FastCRUD
from sqlalchemy import update
from app.models.user import User
from app.schemas.user import (
    UserCreateInternal, UserUpdate, UserUpdateInternal,
    UserDelete, UserRead,
)

class UserCRUD(FastCRUD[
    User,
    UserCreateInternal,
    UserUpdate,
    UserUpdateInternal,
    UserDelete,
    UserRead,
]):
    async def restore(self, *, db, user_id: int) -> UserRead:
        """
        Re-activate a soft-deleted user.
        """
        stmt = (
            update(User)
            .where(User.id == user_id)
            .values(is_deleted=False, deleted_at=None)
            .execution_options(synchronize_session="fetch")
        )
        await db.execute(stmt)
        await db.commit()
        return await self.get(db=db, id=user_id, schema_to_select=UserRead)

crud_users_extended = UserCRUD(User)

The subclass inherits all built-in FastCRUD methods automatically. The restore method uses raw SQLAlchemy for the update operation, then leverages self.get to return a consistently formatted response using the existing UserRead schema.

Working with Specialized Read Schemas

When extending FastCRUD for specific endpoints, you can instantiate separate CRUD objects with alternative read schemas to control data exposure.

Creating Lightweight List Views

Define a minimal schema in src/app/schemas/user_list.py:

from pydantic import BaseModel

class UserListRead(BaseModel):
    id: int
    username: str
    email: str

    class Config:
        orm_mode = True

Then create a specialized CRUD instance in src/app/crud/crud_user_list.py:

from fastcrud import FastCRUD
from app.models.user import User
from app.schemas.user import UserCreateInternal, UserUpdate, UserUpdateInternal, UserDelete
from app.schemas.user_list import UserListRead

UserListCRUD = FastCRUD[
    User,
    UserCreateInternal,
    UserUpdate,
    UserUpdateInternal,
    UserDelete,
    UserListRead,
]
crud_user_list = UserListCRUD(User)

Now crud_user_list.get_multi(..., schema_to_select=UserListRead) returns a lightweight payload ideal for large tables and list endpoints, reducing serialization overhead.

Advanced Query Patterns with Joined Relationships

Extend FastCRUD to encapsulate complex query logic using the built-in get_multi_joined method for eager-loading relationships.

Cross-Model Queries

In src/app/crud/crud_posts.py, subclass FastCRUD to add a method that retrieves posts with author details:

from fastcrud import FastCRUD
from app.models.post import Post
from app.models.user import User
from app.schemas.post import (
    PostCreateInternal, PostDelete, PostRead,
    PostUpdate, PostUpdateInternal,
)
from app.schemas.user import UserRead

class PostCRUD(FastCRUD[
    Post,
    PostCreateInternal,
    PostUpdate,
    PostUpdateInternal,
    PostDelete,
    PostRead,
]):
    async def get_posts_with_authors(self, *, db, limit: int = 20):
        """
        Retrieves posts together with a subset of the author fields.
        """
        return await self.get_multi_joined(
            db=db,
            join_model=User,
            join_on=Post.created_by_user_id == User.id,
            schema_to_select=PostRead,
            join_schema_to_select=UserRead,
            join_prefix="author_",
            limit=limit,
        )

crud_posts_extended = PostCRUD(Post)

This extension leverages join_prefix to namespace the joined user fields (e.g., author_id, author_username) while maintaining type safety through the generic schema definitions.

Integrating Extended CRUD into API Routes

Wire your extended CRUD instances into FastAPI routers by importing them alongside your standard dependency injection patterns:

from fastapi import APIRouter, Depends, HTTPException
from app.db.session import get_db
from app.crud.custom_user import crud_users_extended
from app.schemas.user import UserCreateInternal, UserRead

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserRead)
async def create_user(payload: UserCreateInternal, db=Depends(get_db)):
    return await crud_users_extended.create(db=db, object=payload)

@router.post("/{user_id}/restore", response_model=UserRead)
async def restore_user(user_id: int, db=Depends(get_db)):
    return await crud_users_extended.restore(db=db, user_id=user_id)

This pattern keeps your core boilerplate untouched while allowing unlimited extension of FastCRUD functionalities through subclassing and custom method injection.

Summary

• Subclass the generic CRUD class defined in files like src/app/crud/crud_users.py to add custom business logic while preserving built-in type-safe operations.
• Reuse built-in methods such as self.get and self.update within custom methods to maintain consistent response formatting and reduce code duplication.
• Instantiate multiple CRUD objects with different ReadSchema generics to support varying data exposure requirements for list views versus detail endpoints.
• Leverage get_multi_joined for complex relationship queries, using join_prefix to namespace related model fields automatically.
• Keep extensions modular by storing custom CRUD classes in separate files (e.g., src/app/crud/custom_user.py) and importing them into API routers without modifying the base boilerplate.

Frequently Asked Questions

How do I add a custom filter method to FastCRUD?

Subclass the CRUD class and implement an async method that accepts db: AsyncSession and your filter parameters. Use SQLAlchemy query builders or FastCRUD's filter_criteria parameter to pass arbitrary expressions, then return the results using self.get_multi or direct database execution.

Can I use different Pydantic schemas for the same model in different endpoints?

Yes. Create multiple FastCRUD instantiations with different ReadSchema generics, as shown with UserListRead in src/app/schemas/user_list.py. This allows you to return lightweight schemas for list endpoints and full schemas for detail views without duplicating model logic.

Does extending FastCRUD break the existing type safety?

No. Subclassing preserves all generic type bindings established in the parent class. Your custom methods inherit the same Model, CreateSchema, UpdateSchema, and ReadSchema constraints, ensuring mypy and IDE autocomplete continue to work across extended operations.

How do I handle transactions across multiple CRUD operations in an extended method?

Use the same db: AsyncSession parameter across all operations within your custom method. Since FastCRUD uses the provided session without committing internally (depending on the specific method), you can execute multiple self.update or self.create calls followed by a single await db.commit() to maintain atomic transactions.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}