FastAPI + SQLAlchemy 2.0 Async Engine

Learning outcomes

• Configure async connection pools with asyncpg
• Prevent session leaks across concurrent FastAPI routes

Mental model

SQLAlchemy 2.0 introduces native async support over AsyncIO drivers (asyncpg). Combining FastAPI dependency injection with AsyncSession guarantees clean connection acquisition and teardown on every request tick.

Theory

When running async route handlers (async def), executing blocking synchronous ORM calls (session.query()) blocks Python's single-threaded event loop. AsyncEngine combined with asyncpg offloads database socket I/O to non-blocking epoll events.

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select
from fastapi import FastAPI, Depends

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

engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]

async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    return {"user": user.username if user else None}

Alternatives and trade-offs

• Synchronous ORM (psycopg2): Simple; blocks AsyncIO event loop unless wrapped in run_in_executor.
• SQLAlchemy 2.0 Async (asyncpg): High concurrency non-blocking I/O; requires explicit await on all DB operations and lazy loading handling.

Failure modes and misconceptions

1. Implicit Lazy Loading Crashes: Accessing un-loaded ORM relationships (user.orders) in AsyncIO raises MissingGreenlet error. Use explicit joined loads (options(joinedload(User.orders))).
2. expire_on_commit=True Overhead: Default SQLAlchemy setting expires object attributes on commit, forcing secondary async DB fetches when accessing object attributes after commit. Always set expire_on_commit=False.

Decision scenario

Always set expire_on_commit=False on async_sessionmaker and use explicit eager loading (joinedload/selectinload) to avoid MissingGreenlet errors in FastAPI async routes.

Learning outcomes

• Build non-blocking FastAPI database integration with SQLAlchemy 2.0 and asyncpg.
• Manage AsyncSession lifecycles safely using FastAPI dependency injection.
• Prevent MissingGreenlet errors using explicit eager loading strategies.

Trade-offs

SQLAlchemy 2.0 Async delivers massive concurrency under high HTTP load, but requires disciplined handling of relationship loading and explicit session scopes.

Primary sources

• FastAPI Framework Architecture & Dependency Injection Specification — Tiangolo, verified 2026-07-22