How to Use FastCRUD with SQLAlchemy 2.0 Async Sessions: A Complete Guide

FastCRUD is built natively around SQLAlchemy 2.0's async API, requiring an AsyncSession dependency for all CRUD operations through the FastCRUD class and crud_router helper.

FastCRUD (benavlabs/fastcrud) provides automated CRUD endpoints for FastAPI applications using SQLAlchemy 2.0. Because the library is architected specifically for modern async Python, every operation expects an AsyncSession instance rather than synchronous sessions. This guide demonstrates how to configure SQLAlchemy 2.0 async engines, wire session dependencies into FastCRUD, and build production-ready async CRUD APIs.

Understanding FastCRUD's Async Architecture

The FastCRUD Class and AsyncSession Dependency

At the core of the library is the FastCRUD class defined in fastcrud/crud/fast_crud.py. This generic class accepts a SQLAlchemy model and exposes async methods including create, get, get_multi, update, and delete. Every method signature requires db: AsyncSession as the first parameter, ensuring all database operations are non-blocking.

EndpointCreator and Session Injection

When generating FastAPI routes, the EndpointCreator class in fastcrud/endpoint/endpoint_creator.py injects a session dependency via self.session (lines 307-308). This dependency must be a callable that yields an AsyncSession, which FastCRUD uses to execute queries asynchronously. The _create_item and _read_items methods (lines 604-610 and 66-78 respectively) demonstrate how the session is awaited within each route handler.

Configuring SQLAlchemy 2.0 for FastCRUD

Creating the Async Engine and Sessionmaker

To use FastCRUD with SQLAlchemy 2.0 async sessions, you must configure an async driver. Use create_async_engine and sessionmaker with class_=AsyncSession:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=False)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

FastAPI Dependency for AsyncSession

Create a dependency that yields an AsyncSession for each request. This matches the pattern expected by FastCRUD's EndpointCreator:

from typing import AsyncGenerator
from fastapi import Depends

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

Implementing FastCRUD with Async Sessions

Using crud_router for Auto-Generated Endpoints

The crud_router helper function provides the fastest path to production. Pass your get_session dependency to the session parameter:

from fastapi import FastAPI
from fastcrud import crud_router

# Assuming Item, ItemCreate, ItemUpdate are defined SQLAlchemy/Pydantic models

item_router = crud_router(
    session=get_session,           # AsyncSession dependency

    model=Item,
    create_schema=ItemCreate,
    update_schema=ItemUpdate,
    path="/items",
    tags=["Items"],
)

app = FastAPI()
app.include_router(item_router)

All generated routes (POST /items, GET /items/{id}, GET /items, PATCH /items/{id}, DELETE /items/{id}) automatically await the async session and handle commits internally.

Manual CRUD Operations with FastCRUD Class

For custom business logic, instantiate FastCRUD directly and pass the AsyncSession from your route dependency:

from fastapi import APIRouter, Depends, HTTPException
from fastcrud import FastCRUD
from fastcrud.core.protocols import CRUDInstance
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()

# Instantiate FastCRUD for the Item model

item_crud: CRUDInstance = FastCRUD[Item, ItemCreate, ItemUpdate, ItemUpdate, None, None](
    model=Item,
    is_deleted_column="is_deleted",   # Optional soft-delete config

)

@router.get("/custom/items/{item_id}")
async def read_item(
    item_id: int,
    db: AsyncSession = Depends(get_session),
):
    item = await item_crud.get(
        db, 
        id=item_id, 
        schema_to_select=None, 
        return_as_model=False
    )
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

This approach allows you to mix auto-generated endpoints with hand-crafted logic while reusing the same async CRUD engine.

Advanced Async Features

Pagination and Filtering with Async Sessions

FastCRUD supports offset pagination, cursor pagination, and dynamic filtering through the get_multi method. The EndpointCreator._read_items method (lines 54-58 and 66-78) handles pagination logic, while create_dynamic_filters in fastcrud/fastapi_dependencies.py (lines 27-34) generates query parameter dependencies.

Example request with async filtering:

GET /items?sort=-name&page=2&itemsPerPage=20&tier_id=1

• sort=-name triggers descending order via the async query builder (handled in EndpointCreator._read_items)
• page and itemsPerPage calculate async offset/limit (computed at lines 54-58)
• tier_id=1 applies async filter conditions before returning results

Soft Deletes and Async Transactions

When configured with is_deleted_column, FastCRUD performs async soft deletes by updating the flag rather than executing DELETE statements. All operations respect async session transaction boundaries, ensuring proper await db.commit() and await db.refresh() calls as implemented in fastcrud/endpoint/endpoint_creator.py (lines 20-26).

Summary

• FastCRUD is architected specifically for SQLAlchemy 2.0 async operations, with every CRUD method requiring an AsyncSession parameter.
• Configure your database with create_async_engine and sessionmaker(class_=AsyncSession), then expose a FastAPI dependency that yields sessions using async with.
• Use crud_router for rapid API development by passing your async session dependency, or instantiate FastCRUD directly for custom endpoint logic.
• All pagination, filtering, and soft-delete features operate asynchronously through the same session interface defined in fastcrud/crud/fast_crud.py and fastcrud/endpoint/endpoint_creator.py.

Frequently Asked Questions

Does FastCRUD support synchronous SQLAlchemy sessions?

No. FastCRUD is built exclusively for SQLAlchemy 2.0's async API. The FastCRUD class in fastcrud/crud/fast_crud.py declares all methods as async and expects db: AsyncSession as the first argument. Attempting to pass a synchronous Session will result in type errors and runtime failures because the code internally awaits session methods like await db.execute() and await db.commit().

How does FastCRUD handle database transactions with async sessions?

FastCRUD relies on the standard SQLAlchemy 2.0 async transaction model. Each endpoint awaits session operations such as await db.execute(), await db.commit(), and await db.refresh() as implemented in fastcrud/endpoint/endpoint_creator.py (lines 20-26). The session dependency should use async with async_session() as session to ensure proper transaction scoping, connection pooling, and automatic cleanup when the request completes.

Can I use FastCRUD with databases other than SQLite?

Yes. FastCRUD is database-agnostic regarding SQLAlchemy 2.0 async support. While examples often use sqlite+aiosqlite, you can substitute any async-compatible driver such as postgresql+asyncpg or mysql+aiomysql. Simply update the DATABASE_URL passed to create_async_engine and ensure your sessionmaker uses class_=AsyncSession. FastCRUD's internal queries use standard SQLAlchemy 2.0 async patterns that work across all supported async drivers.

What is the performance benefit of using async sessions with FastCRUD?

Using AsyncSession allows FastCRUD to operate without blocking the FastAPI event loop during database I/O. This enables handling concurrent requests efficiently while waiting for database operations to complete. According to the implementation in fastcrud/crud/fast_crud.py, all methods are coroutines that await the session, allowing the server to process other requests during query execution rather than consuming a thread per request. This architecture significantly improves throughput under high concurrency compared to synchronous alternatives.

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