How to Use FastCRUD for Generating CRUD Operations and Pagination in FastAPI
FastCRUD is a lightweight SQLAlchemy wrapper that auto-generates CRUD methods and provides built-in pagination utilities, eliminating boilerplate code in FastAPI applications.
The benavlabs/fastapi-boilerplate repository implements FastCRUD as its primary data access layer. This pattern allows you to define a single CRUD class per SQLAlchemy model and immediately gain access to standardized create, read, update, delete, and pagination operations without writing repetitive database queries.
What Is FastCRUD?
FastCRUD is a generic type wrapper that binds a SQLAlchemy model to Pydantic schemas for different operations. When instantiated, it exposes methods like create(), get(), get_multi(), update(), delete(), and exists() that handle the underlying database transactions automatically.
In the fastapi-boilerplate, FastCRUD definitions live in src/app/crud/ and follow a consistent naming convention: crud_<model>.py.
Setting Up FastCRUD for Your Model
To implement FastCRUD for a new model, you define a specialized class using the generic FastCRUD type and instantiate it with your SQLAlchemy model.
Create a file at src/app/crud/crud_articles.py:
from fastcrud import FastCRUD
from ..models.article import Article
from ..schemas.article import (
ArticleCreate,
ArticleUpdate,
ArticleUpdateInternal,
ArticleDelete,
ArticleRead,
)
# Define the CRUD class with type parameters
CRUDArticle = FastCRUD[
Article,
ArticleCreate,
ArticleUpdate,
ArticleUpdateInternal,
ArticleDelete,
ArticleRead,
]
# Instantiate the CRUD object
crud_articles = CRUDArticle(Article)
The type parameters map to:
- Model: The SQLAlchemy declarative base class
- CreateSchema: Pydantic model for creation operations
- UpdateSchema: Pydantic model for updates
- UpdateInternalSchema: Internal update schema (often includes auto-generated fields)
- DeleteSchema: Schema for delete operations
- ReadSchema: Pydantic model for read/return operations
Implementing CRUD Operations
Once instantiated, the crud_articles object provides standardized methods for database operations. These methods are async-compatible and work directly with SQLAlchemy's AsyncSession.
Creating Records
Use the create() method to insert new rows. Pass the Pydantic create schema and optionally specify a schema_to_select to return the created record in a specific format:
@router.post("/article", response_model=ArticleRead, status_code=201)
async def create_article(
article: ArticleCreate,
db: AsyncSession = Depends(async_get_db),
):
created = await crud_articles.create(
db=db,
object=article,
schema_to_select=ArticleRead,
)
return created
Reading Records (Single and Multiple)
Fetch a single record using get() with filters:
article = await crud_articles.get(
db=db,
slug=article_slug,
schema_to_select=ArticleRead,
)
For bulk retrieval, use get_multi() with offset and limit parameters:
articles = await crud_articles.get_multi(
db=db,
offset=0,
limit=10,
is_deleted=False,
)
Updating Records
The update() method handles partial updates. Pass the Pydantic update schema and filter criteria to identify the target record:
@router.patch("/article/{slug}")
async def update_article(
slug: str,
values: ArticleUpdate,
db: AsyncSession = Depends(async_get_db),
):
await crud_articles.update(
db=db,
object=values,
slug=slug,
)
return {"message": "Article updated"}
Deleting Records (Soft and Hard)
FastCRUD supports both soft deletes and hard deletes. The delete() method performs a soft delete (setting an is_deleted flag) if your model supports it:
await crud_articles.delete(db=db, slug=slug)
For permanent removal from the database, use db_delete():
await crud_articles.db_delete(db=db, slug=slug)
Adding Pagination to List Endpoints
FastCRUD provides utility functions to standardize pagination across your API. The pattern involves three steps: calculating the offset, fetching the data, and wrapping the response.
Calculating Offsets with compute_offset
The compute_offset() function converts page numbers to SQL offsets:
from fastcrud import compute_offset
offset = compute_offset(page, items_per_page)
# Returns: (page - 1) * items_per_page
Fetching Paginated Data with get_multi
Pass the calculated offset and limit to get_multi():
data = await crud_articles.get_multi(
db=db,
offset=compute_offset(page, items_per_page),
limit=items_per_page,
is_deleted=False,
)
Building the Response with paginated_response
Wrap the results using paginated_response() to generate a standardized structure:
from fastcrud import paginated_response, PaginatedListResponse
return paginated_response(
crud_data=data,
page=page,
items_per_page=items_per_page,
)
The return type is PaginatedListResponse[ArticleRead], which includes total, page, items_per_page, and data fields.
Complete Working Example
Here is a complete router implementation combining CRUD operations and pagination, based on the pattern found in src/app/api/v1/users.py:
from fastapi import APIRouter, Depends, Request
from fastcrud import PaginatedListResponse, compute_offset, paginated_response
from sqlalchemy.ext.asyncio import AsyncSession
from ..core.db.database import async_get_db
from ..crud.crud_articles import crud_articles
from ..schemas.article import ArticleCreate, ArticleRead, ArticleUpdate
router = APIRouter(tags=["articles"])
@router.post("/article", response_model=ArticleRead, status_code=201)
async def create_article(
request: Request,
article: ArticleCreate,
db: AsyncSession = Depends(async_get_db),
):
created = await crud_articles.create(
db=db,
object=article,
schema_to_select=ArticleRead,
)
return created
@router.get("/articles", response_model=PaginatedListResponse[ArticleRead])
async def list_articles(
request: Request,
db: AsyncSession = Depends(async_get_db),
page: int = 1,
items_per_page: int = 10,
):
data = await crud_articles.get_multi(
db=db,
offset=compute_offset(page, items_per_page),
limit=items_per_page,
is_deleted=False,
)
return paginated_response(
crud_data=data,
page=page,
items_per_page=items_per_page,
)
@router.patch("/article/{slug}")
async def update_article(
request: Request,
slug: str,
values: ArticleUpdate,
db: AsyncSession = Depends(async_get_db),
):
await crud_articles.update(db=db, object=values, slug=slug)
return {"message": "Article updated"}
@router.delete("/article/{slug}")
async def delete_article(
request: Request,
slug: str,
db: AsyncSession = Depends(async_get_db),
):
await crud_articles.delete(db=db, slug=slug)
return {"message": "Article deleted"}
Summary
- FastCRUD eliminates repetitive SQLAlchemy boilerplate by auto-generating standard CRUD methods for your models.
- Define a FastCRUD class by specializing the generic
FastCRUDtype with your SQLAlchemy model and Pydantic schemas, then instantiate it as a singleton (e.g.,crud_users). - Available methods include
create(),get(),get_multi(),update(),delete()(soft delete),db_delete()(hard delete), andexists(). - For pagination, use
compute_offset()to calculate SQL offsets,get_multi()withoffsetandlimitparameters to fetch data, andpaginated_response()to wrap results in aPaginatedListResponseschema. - Reference implementations are located in
src/app/crud/crud_users.pyandsrc/app/api/v1/users.pywithin thebenavlabs/fastapi-boilerplaterepository.
Frequently Asked Questions
What is the difference between delete and db_delete in FastCRUD?
The delete() method performs a soft delete by setting an is_deleted flag on the record if your SQLAlchemy model supports soft deletion, keeping the data in the database but filtering it from standard queries. The db_delete() method performs a hard delete, permanently removing the row from the database using SQLAlchemy's delete() operation. Choose delete() for data integrity and audit trails, or db_delete() for permanent removal.
How does FastCRUD handle pagination limits?
FastCRUD handles pagination through the get_multi() method's offset and limit parameters. The helper function compute_offset(page, items_per_page) calculates the SQL offset as (page - 1) * items_per_page. You pass this offset and your desired page size to get_multi(), which applies these values directly to the SQLAlchemy query using offset() and limit() clauses. The paginated_response() utility then wraps the results with metadata including total count, current page, and items per page.
Can I use FastCRUD with async SQLAlchemy?
Yes, FastCRUD is fully compatible with async SQLAlchemy. All generated methods such as create(), get(), get_multi(), update(), and delete() are designed to work with AsyncSession from sqlalchemy.ext.asyncio. The methods are awaitable and handle async database operations internally. When using FastCRUD in the fastapi-boilerplate, you inject the async database session using Depends(async_get_db) as shown in the router implementations in src/app/api/v1/users.py.
Where are the CRUD definitions stored in the fastapi-boilerplate?
CRUD definitions are stored in the src/app/crud/ directory. Each SQLAlchemy model has a corresponding file named crud_<model>.py (e.g., crud_users.py, crud_tier.py, crud_posts.py). These files contain the FastCRUD class specialization and the singleton instance used throughout the application. The instantiated CRUD objects are then imported by API routers in src/app/api/v1/ to handle endpoint logic.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →