# Python Sqlalchemy

> SQLAlchemy 2.0 stable ORM and Core patterns. Use for typed mappings, select() queries, sessions and transaction boundaries, relationships and loading, sync or async engines, JSON, dialect-specific upserts, or Alembic migrations. For HTTP concerns, use python-fastapi.

- Skill: `martinffx/python-sqlalchemy` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add martinffx/python-sqlalchemy`
- Raw SKILL.md: https://api.skillmd.com/api/skills/martinffx/python-sqlalchemy/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: martinffx (https://skillmd.com/u/martinffx)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/martinffx/python-sqlalchemy

---


# SQLAlchemy ORM Patterns

Use SQLAlchemy 2.0 stable APIs. `Session.query()` is a legacy compatibility API; new code uses `select()`.

## Model Definition

```python
from decimal import Decimal
from uuid import UUID

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

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+psycopg://user:pass@localhost/db")
SessionLocal = sessionmaker(bind=engine)

def get_db():
    with SessionLocal() as session:
        yield session
```

The dependency owns closure only. The application use case owns `with session.begin():` or explicit commit/rollback. Repositories flush or add records; they do not commit each operation.

## Query Patterns

```python
from sqlalchemy import func, select

products = session.scalars(
    select(ProductModel).where(ProductModel.in_stock)
).all()

product = session.get(ProductModel, product_id)
count = session.scalar(select(func.count()).select_from(ProductModel))
```

## Upsert

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

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

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

product = session.scalars(
    stmt.returning(ProductModel),
    execution_options={"populate_existing": True},
).one()
```

This is PostgreSQL-specific. The conflict target must be backed by a unique constraint or index. Keep transaction ownership outside the repository.

## Relationships

```python
from sqlalchemy import ForeignKey
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 typing import Any

from sqlalchemy import JSON, select
from sqlalchemy.ext.mutable import MutableDict

class ConfigModel(Base):
    __tablename__ = "configs"
    id: Mapped[int] = mapped_column(primary_key=True)
    settings: Mapped[dict[str, Any]] = mapped_column(MutableDict.as_mutable(JSON))

configs = session.scalars(
    select(ConfigModel).where(ConfigModel.settings["theme"].as_string() == "dark")
).all()
```

Plain `JSON` does not detect in-place dict mutations. Use replacement assignment or `MutableDict`; nested values need their own mutation policy.

## References

- [Models](references/models.md): mappings, defaults, relationships, and JSON types.
- [Queries](references/queries.md): results, loading, pagination, and large result sets.
- [Async](references/async.md): sessions per task, implicit I/O, and disposal.

Changing ORM metadata does not migrate existing databases. Use Alembic migration scripts and review autogenerated revisions, especially destructive changes and data backfills.