Async Database

Difficulty: Advanced

Async database access lets FastAPI handle database queries without blocking the event loop, enabling much higher concurrency for I/O-bound applications. SQLAlchemy 2.0 provides full async support through its asyncio extension, working with async database drivers like asyncpg (PostgreSQL), aiosqlite (SQLite), and aiomysql (MySQL).

The async setup mirrors the synchronous setup but uses async-specific classes. Instead of create_engine, use create_async_engine. Instead of sessionmaker, configure it with class_=AsyncSession. Instead of Session, use AsyncSession. The database URL uses an async driver (postgresql+asyncpg:// instead of postgresql://).

Async queries use the new select() syntax instead of the older query() API. Instead of db.query(User).filter(...).all(), you write result = await db.execute(select(User).where(...)) followed by result.scalars().all(). This explicit execute-then-extract pattern works well with async/await.

The async session dependency uses an async generator (async def with yield). The async with statement manages the session lifecycle. You can commit on success and rollback on failure within the same dependency.

Async database access shines when your API handles many concurrent requests that each perform database I/O. While one request waits for a database query, the event loop can process other requests. This is particularly beneficial for applications that make multiple database calls per request or call external APIs alongside database operations.

Note that async database access is not always faster for individual requests. The overhead of async can make single requests slightly slower. The benefit is in concurrency - handling many simultaneous requests efficiently. If your API has low concurrency, synchronous database access is simpler and equally effective.

Code examples

Async SQLAlchemy Setup

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from sqlalchemy import String

# Async engine (note the async driver)
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/mydb"
# For SQLite: "sqlite+aiosqlite:///./app.db"

async_engine = create_async_engine(DATABASE_URL, echo=True)

# Async session factory
AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

# Async session dependency
async def get_async_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

create_async_engine uses an async driver (asyncpg/aiosqlite). The session factory creates AsyncSession instances. expire_on_commit=False prevents lazy loading issues. The dependency commits on success and rolls back on failure.

Async CRUD Operations

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pydantic import BaseModel, ConfigDict

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    email: str

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_async_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    await db.flush()  # Flush to get the ID
    await db.refresh(db_user)
    return db_user

@app.get("/users", response_model=list[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 20,
    db: AsyncSession = Depends(get_async_db),
):
    result = await db.execute(
        select(User).offset(skip).limit(limit)
    )
    return result.scalars().all()

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Async endpoints use async def and await for database operations. db.execute() runs the query, and result.scalars().all() extracts the objects. scalar_one_or_none() returns a single result or None. flush() persists without committing (commit happens in the dependency).

Async Filtering and Relationships

from sqlalchemy import select, or_, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

async def search_users(
    db: AsyncSession,
    q: str | None = None,
    is_active: bool | None = None,
):
    stmt = select(User)

    if q:
        stmt = stmt.where(
            or_(
                User.name.ilike(f"%{q}%"),
                User.email.ilike(f"%{q}%"),
            )
        )

    if is_active is not None:
        stmt = stmt.where(User.is_active == is_active)

    result = await db.execute(stmt.order_by(User.name))
    return result.scalars().all()

# Count records
async def count_users(db: AsyncSession) -> int:
    result = await db.execute(select(func.count()).select_from(User))
    return result.scalar_one()

# Eager load relationships
async def get_user_with_posts(db: AsyncSession, user_id: int):
    result = await db.execute(
        select(User)
        .where(User.id == user_id)
        .options(selectinload(User.posts))
    )
    return result.scalar_one_or_none()

Async queries use select() with .where() for filtering. func.count() counts records. selectinload() eagerly loads relationships to avoid lazy loading issues in async contexts (lazy loading does not work with async).

Key points

Concepts covered

AsyncSession, asyncpg, async Engine, select, Async CRUD