Async SQLAlchemy Patterns Skill

Production-ready async SQLAlchemy 2.0+ patterns for FastAPI applications.

Overview

This skill provides comprehensive guidance and tooling for implementing async SQLAlchemy in FastAPI projects, including:

Async session management with proper connection pooling
Alembic migrations setup and best practices
Model patterns with relationships, indexes, and constraints
Query optimization techniques and anti-patterns
Performance monitoring and troubleshooting

Quick Start

1. Setup Database Session

Use the session manager template:

from templates.session_manager import get_db

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

2. Create Models

Use the base model template:

from templates.base_model import BaseModel

class User(BaseModel):
    __tablename__ = "users"

    email: Mapped[str] = mapped_column(String(255), unique=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

3. Setup Migrations

./scripts/setup-alembic.sh
./scripts/generate-migration.sh "initial schema"
alembic upgrade head

Key Files

Scripts

setup-alembic.sh - Initialize Alembic with async support
generate-migration.sh - Create migrations from model changes
test-connection.sh - Test database connectivity
optimize-queries.sh - Analyze queries for optimization

Templates

base_model.py - Base model with UUID, timestamps, soft delete
session_manager.py - Async session factory and dependencies
alembic.ini - Production-ready Alembic configuration
env.py - Alembic environment with async support

Examples

user_model.py - Complete model with relationships and indexes
async_context_examples.py - 12 session usage patterns
query_patterns.py - Advanced query optimization examples

Common Patterns

Session Management

# FastAPI dependency (recommended)
async def endpoint(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

# Manual context manager
async with AsyncSessionLocal() as session:
    user = User(email="test@example.com")
    session.add(user)
    await session.commit()

Eager Loading

# Avoid N+1 queries with selectinload
stmt = select(User).options(
    selectinload(User.posts),
    joinedload(User.roles)
)
result = await session.execute(stmt)
users = result.scalars().unique().all()  # unique() with joins!

Bulk Operations

# Bulk insert
users = [User(**data) for data in users_data]
session.add_all(users)
await session.commit()

# Bulk update with RETURNING
stmt = (
    update(User)
    .where(User.id.in_(user_ids))
    .values(is_active=True)
    .returning(User)
)
result = await session.execute(stmt)
await session.commit()

Transactions

async with session.begin_nested():  # Savepoint
    # Operations here
    await session.execute(stmt1)
    await session.execute(stmt2)
    # Auto-rollback on exception

await session.commit()

Performance Tips

Always use eager loading for relationships to avoid N+1 queries
Enable SQL echo during development to see generated queries
Add indexes for frequently filtered/sorted columns
Use bulk operations for large datasets
Configure connection pooling appropriately for your workload
Stream large results instead of loading all at once

Troubleshooting

Common Issues

"Object is not bound to session"

# Fix: Configure session with expire_on_commit=False
AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,
)

"DetachedInstanceError"

# Fix: Use eager loading within session context
stmt = select(User).options(selectinload(User.posts))

"Connection pool exhausted"

# Fix: Increase pool size
engine = create_async_engine(
    url,
    pool_size=20,
    max_overflow=40,
)

Best Practices

✅ Always review auto-generated migrations before applying
✅ Use type hints with Mapped[] for all columns
✅ Add indexes for foreign keys and frequently queried columns
✅ Configure proper cascade rules for relationships
✅ Use soft delete instead of hard delete when possible
✅ Enable connection health checks with pool_pre_ping=True
✅ Set appropriate pool_recycle time for your database

Database Support

PostgreSQL (recommended): postgresql+asyncpg://
MySQL: mysql+aiomysql://
SQLite: sqlite+aiosqlite://

Version Compatibility

SQLAlchemy: 2.0+
Python: 3.10+
FastAPI: 0.100+
Alembic: 1.12+

Resources

License

Part of the fastapi-backend plugin for Claude Code.

async-sqlalchemy-patterns