How to use SQLAlchemy async sessions with FastAPI

Async SQLAlchemy with FastAPI

SQLAlchemy 1.4+ supports async sessions via create_async_engine. This unlocks true non-blocking database queries in async def FastAPI routes.

Setup

pip install sqlalchemy[asyncio] asyncpg

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

Async yield dependency

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Async CRUD

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(
        select(models.User).where(models.User.id == user_id)
    )
    return result.scalars().first()

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await get_user(db, user_id)
    if not user:
        raise HTTPException(404, "Not found")
    return user

Use the select() construct instead of session.query() — the query API does not support async. scalars().first() extracts the ORM object from the result set. Set expire_on_commit=False to access model attributes after committing without a second query.