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:

from 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:

from 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

Option	Purpose	Default
echo	Log all SQL statements	False
pool_size	Number of persistent connections	5
max_overflow	Additional connections allowed	10
pool_pre_ping	Test connection validity	True
pool_recycle	Recycle connections after N seconds	-1 (disabled)

Session Options

Option	Purpose	Recommended
expire_on_commit	Expire objects after commit	False
autocommit	Auto-commit transactions	False
autoflush	Auto-flush before queries	False

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.

# 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:

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

Examples:

# 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:

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

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

Health Check Example

from 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:

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