SQLAlchemy ORM Patterns

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

Model Definition

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

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

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

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

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

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.