---

name: fastapi-database-setup description: Configure async SQLAlchemy 2.0 engine, session factory, and database dependency for FastAPI

FastAPI Database Setup

Overview

This skill covers setting up async SQLAlchemy 2.0 with PostgreSQL using asyncpg driver.

Create database.py

Create src/app/database.py:

Copyfrom collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import settings

# Create async engine
engine = create_async_engine(
    str(settings.database_url),
    echo=settings.database_echo,
    pool_size=settings.database_pool_size,
    max_overflow=settings.database_max_overflow,
    pool_pre_ping=True,  # Verify connections before use
)

# Create async session factory
async_session_factory = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Don't expire objects after commit
    autocommit=False,
    autoflush=False,
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """
    Dependency that provides a database session.
    
    Yields an async session and ensures it's closed after use.
    Uses the same session throughout the request lifecycle.
    """
    async with async_session_factory() as session:
        try:
            yield session
        finally:
            await session.close()

Create dependencies.py

Create src/app/dependencies.py:

Copyfrom collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession

from app.database import async_session_factory


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """
    Dependency that provides a database session.
    
    Usage in routers:
        @router.get("/items")
        async def list_items(db: AsyncSession = Depends(get_db)):
            ...
    """
    async with async_session_factory() as session:
        try:
            yield session
        finally:
            await session.close()

Key Configuration Options

Engine Options

Session Options

Why expire_on_commit=False?

In async contexts, accessing expired attributes triggers lazy loading, which fails outside an active session. Setting expire_on_commit=False allows accessing model attributes after commit without additional queries.

Copy# With expire_on_commit=True (default) - FAILS
async with session.begin():
    item = Item(name="test")
    session.add(item)
# item.name would fail here - object expired

# With expire_on_commit=False - WORKS
async with session.begin():
    item = Item(name="test")
    session.add(item)
# item.name works - object not expired

Why autoflush=False?

Manual control over when changes are flushed to the database:

• Prevents unexpected queries during read operations
• Gives explicit control over transaction boundaries
• Avoids issues with lazy loading in async context

Connection URL Format

PostgreSQL async connection string format:

Copypostgresql+asyncpg://user:password@host:port/database

Examples:

Copy# Local development
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/myapp

# With SSL
DATABASE_URL=postgresql+asyncpg://user:pass@host:5432/db?ssl=require

# Connection pooler (e.g., PgBouncer)
DATABASE_URL=postgresql+asyncpg://user:pass@pgbouncer:6432/db?prepared_statement_cache_size=0

Testing Configuration

For tests, use a separate test database:

Copy# In config.py, add:
test_database_url: PostgresDsn | None = None

Copy# In .env.example, add:
TEST_DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname_test

Health Check Example

Copyfrom sqlalchemy import text

async def check_database_health(session: AsyncSession) -> bool:
    """Check if database is accessible."""
    try:
        await session.execute(text("SELECT 1"))
        return True
    except Exception:
        return False

Shutdown Cleanup

In your app factory, dispose the engine on shutdown:

Copy@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()