FastAPI Async SQLAlchemy: Alembic Migrations That Survive Production

# io.thecodeforge — Python tutorial

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

# Single engine per app — created once at startup
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,          # Max connections in pool
    max_overflow=10,       # Extra connections if pool exhausted
    pool_pre_ping=True,    # Verify connections before use
    echo=False,            # Set True for debugging SQL
)

# Session factory — thread-safe, creates new sessions per request
async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Avoid expired attribute access after commit
)

class Base(DeclarativeBase):
    pass

# io.thecodeforge — Python tutorial

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_session() -> AsyncSession:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()  # Commit if no exception
        except Exception:
            await session.rollback()  # Rollback on error
            raise  # Re-raise so FastAPI handles it
        # Session closed automatically by async with

@app.post("/orders/")
async def create_order(session: AsyncSession = Depends(get_session)):
    # Use session — no need to close manually
    result = await session.execute(...)
    return {"status": "ok"}

# io.thecodeforge — Python tutorial

from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from alembic import context

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

from app.models import Base  # your declarative base
target_metadata = Base.metadata

# Use the same async engine as your app
async_engine: AsyncEngine = create_async_engine(
    config.get_main_option("sqlalchemy.url"),
    poolclass=pool.NullPool,  # No pooling for migrations
)

def run_migrations_offline() -> None:
    context.configure(url=async_engine.url, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    async with async_engine.connect() as conn:
        await conn.run_sync(do_run_migrations)
    await async_engine.dispose()

def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    import asyncio
    asyncio.run(run_async_migrations())

# Choose mode based on context
if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

# io.thecodeforge — Python tutorial

"""Add is_active column to users table."""
from alembic import op
import sqlalchemy as sa

revision = 'xxxx'
down_revision = 'yyyy'

def upgrade():
    # Set lock timeout to avoid blocking production
    op.execute('SET lock_timeout = 5000')
    
    # Add column as nullable first
    op.add_column('users', sa.Column('is_active', sa.Boolean(), nullable=True))
    
    # Backfill data in batches (run separately in low traffic)
    # This is a placeholder — actual backfill should be a separate script
    
    # Then set NOT NULL if needed
    op.alter_column('users', 'is_active', nullable=False, server_default=sa.text('true'))

def downgrade():
    op.drop_column('users', 'is_active')

# io.thecodeforge — Python tutorial

import subprocess
import psycopg2
from sqlalchemy import create_engine

def test_migrations():
    # Assume a test DB is already created and populated
    # Run migrations
    result = subprocess.run(
        ["alembic", "upgrade", "head"],
        capture_output=True, text=True
    )
    assert result.returncode == 0, f"Migration failed: {result.stderr}"
    
    # Verify schema by inspecting tables
    engine = create_engine("postgresql://user:pass@localhost/testdb")
    with engine.connect() as conn:
        # Check that expected columns exist
        result = conn.execute("SELECT column_name FROM information_schema.columns WHERE table_name='users'")
        columns = [row[0] for row in result]
        assert 'is_active' in columns, "Column is_active missing after migration"
    
    # Test rollback
    result = subprocess.run(
        ["alembic", "downgrade", "-1"],
        capture_output=True, text=True
    )
    assert result.returncode == 0, f"Rollback failed: {result.stderr}"
    print("All migration tests passed.")

if __name__ == "__main__":
    test_migrations()

# io.thecodeforge — Python tutorial

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=16,          # Adjust based on DB limits
    max_overflow=10,       # Allow bursts
    pool_pre_ping=True,    # Check connection health before use
    pool_recycle=240,      # Recycle connections every 4 minutes
    pool_use_lifo=True,    # Use LIFO to reduce connection churn
)

# io.thecodeforge — Python tutorial

# Use async SQLAlchemy when:
# - High concurrency (> 1000 concurrent requests)
# - Many I/O calls per request (DB + external APIs)
# - Real-time features (WebSockets, streaming)
#
# Use sync SQLAlchemy when:
# - Low concurrency (< 100 req/s)
# - Simple CRUD with minimal I/O
# - Team is not comfortable with async debugging
#
# Example: a dashboard that fetches 10 DB queries per request
# Async can run them concurrently, cutting response time from 1s to 200ms.

# io.thecodeforge — Python tutorial

import logging
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from sqlalchemy.orm import sessionmaker

logger = logging.getLogger(__name__)

class LoggedAsyncSession(AsyncSession):
    async def close(self):
        logger.debug(f"Closing session {id(self)}")
        await super().close()

# Use this factory to track sessions
async_session_factory = async_sessionmaker(
    engine,
    class_=LoggedAsyncSession,
    expire_on_commit=False,
)

# In a health endpoint:
@app.get("/health")
async def health():
    # Check pool status
    pool = engine.pool
    return {
        "pool_size": pool.size(),
        "checked_in_connections": pool.checkedin(),
        "overflow": pool.overflow(),
    }