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
```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
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: mappings, defaults, relationships, and JSON types.
- Queries: results, loading, pagination, and large result sets.
- Async: 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.