home / skills / martinffx / claude-code-atelier / sqlalchemy

sqlalchemy skill

/plugins/atelier-python/skills/sqlalchemy

This skill helps you implement robust SQLAlchemy ORM patterns in Python, covering models, sessions, queries, upserts, relationships, and JSON fields.

This is most likely a fork of the atelier-python-sqlalchemy skill from martinffx
npx playbooks add skill martinffx/claude-code-atelier --skill sqlalchemy

Review the files below or copy the command above to add this skill to your agents.

Files (4)
SKILL.md
2.7 KB
---
name: python:sqlalchemy
description: SQLAlchemy ORM patterns for Python database access. Use when defining models, writing queries, implementing upserts, working with JSON columns, or managing database sessions.
user-invocable: false
---

# SQLAlchemy ORM Patterns

Modern SQLAlchemy 2.0+ patterns for database access in Python applications.

## Model Definition

```python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String
from uuid import UUID
from decimal import Decimal

class Base(DeclarativeBase):
    pass

class ProductModel(Base):
    __tablename__ = "products"

    id: Mapped[UUID] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[Decimal]
    in_stock: Mapped[bool] = mapped_column(default=True)
```

## Session Management

```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql://user:pass@localhost/db")
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
```

## Query Patterns

```python
# Select
stmt = select(ProductModel).where(ProductModel.price > 100)
products = session.execute(stmt).scalars().all()

# Filter
products = session.query(ProductModel).filter(ProductModel.in_stock == True).all()

# Get by ID
product = session.get(ProductModel, product_id)

# Count
count = session.query(ProductModel).count()
```

## Upsert

```python
from sqlalchemy.dialects.postgresql import insert

stmt = insert(ProductModel).values(
    id=product_id,
    name="Widget",
    price=9.99,
)

# On conflict, update
stmt = stmt.on_conflict_do_update(
    index_elements=["id"],
    set_={"name": stmt.excluded.name, "price": stmt.excluded.price},
)

session.execute(stmt)
session.commit()
```

## Relationships

```python
from sqlalchemy.orm import relationship

class UserModel(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    orders: Mapped[list["OrderModel"]] = relationship(back_populates="user")

class OrderModel(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["UserModel"] = relationship(back_populates="orders")
```

## JSON Columns

```python
from sqlalchemy import JSON

class ConfigModel(Base):
    __tablename__ = "configs"
    id: Mapped[int] = mapped_column(primary_key=True)
    settings: Mapped[dict] = mapped_column(JSON)

# Query JSON field
configs = session.query(ConfigModel).filter(
    ConfigModel.settings["theme"] == "dark"
).all()
```

See references/ for model patterns, query optimization, and async SQLAlchemy.

Overview

This skill provides modern SQLAlchemy 2.0+ ORM patterns for Python database access, focused on clear model definitions, safe session handling, efficient queries, upserts, relationships, and JSON column usage. It packages concise, practical examples you can copy into applications to define models, run queries, and manage transactions reliably.

How this skill works

The skill shows DeclarativeBase model patterns with annotated Mapped columns and typed fields, plus session factory usage for deterministic connection management. It demonstrates common query idioms (select, filter, get, count), PostgreSQL upserts via insert().on_conflict_do_update, relationship setup with relationship/back_populates, and JSON column queries. Example snippets include synchronous session usage and guidance applicable to async variants.

When to use it

  • Defining typed ORM models with SQLAlchemy 2.0+ and Python type hints
  • Implementing safe session lifecycles and dependency-injected DB access
  • Writing common queries: selects, filters, counts, and single-object fetches
  • Performing upserts with PostgreSQL ON CONFLICT semantics
  • Modeling relationships and querying JSON columns in the database

Best practices

  • Use DeclarativeBase and annotated Mapped types to keep models explicit and type-checkable
  • Create a sessionmaker-bound factory and always close sessions in a finally block or use context managers
  • Prefer select() + session.execute(...).scalars() for predictable results in SQLAlchemy 2.0
  • Use database-native upsert (ON CONFLICT) for idempotent writes instead of read-then-write
  • Keep JSON columns for flexible, denormalized data but index and validate fields used in filters

Example use cases

  • Define a ProductModel with UUID id, Decimal price, and default flags for inventory
  • Create a dependency that yields a DB session for use in request handlers or background tasks
  • Run a price-based query to fetch products above a threshold and paginate results
  • Perform an upsert to insert or update product rows in a bulk import job
  • Query user preferences stored in a JSON column to filter by theme or feature flags

FAQ

Can I use these patterns with async SQLAlchemy?

Yes. The concepts carry over; use AsyncEngine, async_sessionmaker, and await session.execute() with async context managers.

How do I handle transactions across multiple operations?

Use session.begin() or context-managed transactions to ensure commit/rollback boundaries and keep transactions short-lived.