Python FastAPI + SQLAlchemy Async Database Connection Pattern

Problem

Using SQLAlchemy synchronously in FastAPI blocks the event loop during database queries. As concurrent requests increase, response times tank.

Solution

Set up async connections with sqlalchemy[asyncio] and asyncpg.

pip install sqlalchemy[asyncio] asyncpg

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

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

engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

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

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy import select
from database import get_db

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_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)
    return user

Key Points

• asyncpg is PostgreSQL-only. Use aiomysql for MySQL or aiosqlite for SQLite.
• Without expire_on_commit=False, accessing object attributes after commit triggers lazy loading, which raises errors in async contexts.
• Tune pool_size and max_overflow based on your concurrency needs. Make sure not to exceed the database’s connection limit.
• Making get_db a generator lets FastAPI’s dependency injection handle session creation and cleanup automatically. No manual try/finally needed.