Claude Dev Suite Claude Dev Suite Sqlalchemy

SQLAlchemy Core Knowledge

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: sqlalchemy for comprehensive documentation.

When NOT to Use This Skill

• TypeScript/Node.js Projects: Use prisma, drizzle, or typeorm skills
• Django Applications: Use Django ORM documentation (not covered here)
• Raw SQL Queries: Use database-query MCP server for direct SQL execution
• NoSQL Databases: Use mongodb skill for MongoDB operations
• FastAPI Specific: May need fastapi-expert for integration patterns
• Database Design: Consult sql-expert or architect-expert for schema architecture
• Other Python ORMs: Peewee, Pony ORM, Tortoise not supported by this skill

Model Definition

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    content = Column(String)
    author_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

Session Operations

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)

# Create
with Session() as session:
    user = User(name='John', email='john@example.com')
    session.add(user)
    session.commit()

# Read
with Session() as session:
    users = session.query(User).all()
    user = session.query(User).filter_by(id=1).first()
    active_users = session.query(User).filter(User.is_active == True).all()

# Update
with Session() as session:
    user = session.query(User).filter_by(id=1).first()
    user.name = 'Jane'
    session.commit()

# Delete
with Session() as session:
    session.query(User).filter_by(id=1).delete()
    session.commit()

Async Support

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db')
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_users():
    async with async_session() as session:
        result = await session.execute(select(User))
        return result.scalars().all()

Anti-Patterns

Quick Troubleshooting

Production Readiness

Engine Configuration

# database.py
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool
import os

DATABASE_URL = os.environ['DATABASE_URL']

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,  # Recycle connections after 30 minutes
    pool_pre_ping=True,  # Check connection validity
    echo=os.environ.get('DEBUG') == 'true',
    connect_args={
        'sslmode': 'require' if os.environ.get('ENV') == 'production' else 'prefer',
        'connect_timeout': 10,
    }
)

# Event listeners for monitoring
@event.listens_for(engine, 'before_cursor_execute')
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(engine, 'after_cursor_execute')
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop()
    if total > 0.5:  # Log slow queries
        logger.warning(f'Slow query ({total:.2f}s): {statement[:100]}')

SessionLocal = sessionmaker(bind=engine)
ScopedSession = scoped_session(SessionLocal)

Context Manager Pattern

from contextlib import contextmanager
from typing import Generator

@contextmanager
def get_db() -> Generator[Session, None, None]:
    """Database session context manager with automatic cleanup."""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
def create_user(name: str, email: str) -> User:
    with get_db() as db:
        user = User(name=name, email=email)
        db.add(user)
        db.flush()  # Get ID without committing
        return user

Async Configuration

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

async_engine = create_async_engine(
    DATABASE_URL.replace('postgresql://', 'postgresql+asyncpg://'),
    pool_size=20,
    max_overflow=10,
    pool_recycle=1800,
    echo=False,
)

async_session = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

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

Query Optimization

from sqlalchemy.orm import joinedload, selectinload
from sqlalchemy import select

# Eager loading to avoid N+1
async def get_users_with_posts(db: AsyncSession):
    stmt = (
        select(User)
        .options(selectinload(User.posts))
        .where(User.is_active == True)
    )
    result = await db.execute(stmt)
    return result.scalars().all()

# Pagination
async def paginate(
    db: AsyncSession,
    page: int,
    per_page: int
) -> dict:
    offset = (page - 1) * per_page

    # Count
    count_stmt = select(func.count(User.id))
    total = await db.scalar(count_stmt)

    # Data
    stmt = (
        select(User)
        .order_by(User.created_at.desc())
        .offset(offset)
        .limit(per_page)
    )
    result = await db.execute(stmt)

    return {
        'data': result.scalars().all(),
        'total': total,
        'page': page,
        'pages': math.ceil(total / per_page),
    }

# Bulk insert
def bulk_insert_users(db: Session, users_data: list[dict]):
    db.execute(insert(User), users_data)
    db.commit()

Alembic Migrations

# alembic/env.py
from sqlalchemy import engine_from_config, pool
from alembic import context
from models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
        )

        with context.begin_transaction():
            context.run_migrations()

# CLI commands
# alembic revision --autogenerate -m "Add users table"
# alembic upgrade head
# alembic downgrade -1

Testing

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope='session')
def engine():
    return create_engine(os.environ['TEST_DATABASE_URL'])

@pytest.fixture(scope='session')
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db(engine, tables):
    connection = engine.connect()
    transaction = connection.begin()
    Session = sessionmaker(bind=connection)
    session = Session()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

# tests/test_user.py
def test_create_user(db):
    user = User(name='Test', email='test@example.com')
    db.add(user)
    db.flush()
    assert user.id is not None

Monitoring Metrics

Checklist

• Connection pooling configured
• SSL in production
• pool_pre_ping enabled
• pool_recycle for long-running apps
• Context manager for sessions
• Eager loading to prevent N+1
• Pagination for list queries
• Alembic migrations
• Slow query logging
• Test isolation with rollback

Reference Documentation

• Relationships
• Async

Source: claude-dev-suite/claude-dev-suite — distributed by TomeVault.