sqlalchemy

What this skill does

# 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

```python
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

```python
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='[email protected]')
    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

```python
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

| Anti-Pattern | Why It's Bad | Better Approach |
|-------------|--------------|------------------|
| Not closing sessions | Connection leaks, pool exhaustion | Use context managers (`with Session()`) |
| Using `session.query()` in SQLAlchemy 2.0+ | Deprecated, legacy API | Use `select()` with `session.execute()` |
| No eager loading for relations | N+1 query problem | Use `selectinload()` or `joinedload()` |
| Hardcoded connection strings | Security risk | Use environment variables |
| Not configuring connection pool | Connection issues, performance | Set `pool_size`, `max_overflow`, `pool_recycle` |
| Using `session.flush()` without understanding | Partial commits, confusion | Use `commit()` for transactions |
| No `pool_pre_ping` in production | Stale connections after DB restart | Enable `pool_pre_ping=True` |
| Missing indexes on foreign keys | Slow joins | Add `Index()` to frequently joined columns |
| Using ORM for bulk operations | Very slow for large datasets | Use `bulk_insert_mappings()` or Core |
| Not handling `IntegrityError` | Cryptic errors to users | Catch and provide meaningful messages |

## Quick Troubleshooting

| Issue | Likely Cause | Solution |
|-------|--------------|----------|
| "No module named 'sqlalchemy'" | SQLAlchemy not installed | Run `pip install sqlalchemy` |
| "Can't connect to server" | Wrong DATABASE_URL or DB down | Verify connection string, check DB status |
| "Table doesn't exist" | Migrations not run | Execute `alembic upgrade head` |
| "DetachedInstanceError" | Accessing relation outside session | Use `joinedload()` or keep session open |
| "IntegrityError: duplicate key" | Unique constraint violation | Check for existing record, handle error |
| "InvalidRequestError: SQL expression" | Missing `select()` in 2.0 style | Use `select(Model)` not `session.query()` |
| Slow queries | N+1 problem, missing indexes | Add eager loading, create indexes |
| Pool timeout | Too many connections | Increase `pool_size`, fix connection leaks |
| "Stale data" | Session caching old data | Call `session.expire_all()` or refresh objects |
| Alembic conflict | Multiple migration heads | Merge branches with `alembic merge` |

## Production Readiness

### Engine Configuration

```python
# 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

```python
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

```python
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

```python
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(selecti