Intermediate Examples

# pyright: strict
"""Example 29: Session Object States -- Transient, Pending, Persistent, Detached."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import Engine, create_engine, inspect, text  # => co-11: inspect() exposes an object's InstanceState
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from sqlalchemy.orm import InstanceState  # => the type inspect() returns for a mapped instance -- carries the phase flags
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-11: the mapped class this example tracks across all four states
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres, None before a flush
    name: Mapped[str]  # => a required TEXT column
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE customer from Customer's Mapped[] fields
 
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table
 
    ada = Customer(name="Ada")  # => co-11: STATE 1 -- TRANSIENT. constructed but no Session has ever seen it
    state: InstanceState[Customer] = inspect(ada)  # => a handle onto the object's own InstanceState -- reads phase flags
    print(f"transient: {state.transient}")  # => Output: transient: True
    assert state.transient and not state.pending and not state.persistent and not state.detached  # => exactly one flag set
 
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        session.add(ada)  # => co-11: STATE 2 -- PENDING. the session knows about `ada`, but no INSERT has been sent yet
        print(f"pending: {state.pending}")  # => Output: pending: True
        assert state.pending and not state.transient and not state.persistent  # => the state object updates itself in place
 
        session.commit()  # => co-11 + co-12: STATE 3 -- PERSISTENT. flushes the INSERT, then commits -- now tracked in the
        # => Session's identity map, with a real primary key backing it
        print(f"persistent: {state.persistent}")  # => Output: persistent: True
        assert state.persistent and not state.pending  # => moved out of "pending" the moment the flush succeeded
        ada_id = ada.id  # => reads `id` INSIDE the still-open session -- avoids a DetachedInstanceError below
 
    print(f"detached: {state.detached}")  # => Output: detached: True
    # => co-11: STATE 4 -- DETACHED. the `with` block above closed the session -- `ada` still exists in Python memory,
    # => still has its `id`, but is no longer tracked by ANY session's identity map or unit of work
    assert state.detached and not state.persistent  # => co-11: the full transient -> pending -> persistent -> detached arc
    print(f"final id={ada_id}")  # => Output: final id=1
    # => co-11: these four states are exactly the four SQLAlchemy uses to describe an object's persistence lifecycle
    print("ex-29 OK")  # => Output: ex-29 OK

transient: True
pending: True
persistent: True
detached: True
final id=1
ex-29 OK

# pyright: strict
"""Example 30: session.expire() + refresh() -- Forcing a Reload From Postgres."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import Engine, create_engine, text  # => co-11: the engine a Session opens connections FROM
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-11: the mapped class this example expires and refreshes
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column, mutated OUTSIDE the ORM later in this example
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE customer from Customer's Mapped[] fields
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        ada = Customer(name="Ada")  # => the row this example mutates OUTSIDE the ORM's own knowledge
        session.add(ada)  # => registers `ada` as pending
        session.commit()  # => flushes the INSERT, assigns ada.id
        ada_id = ada.id  # => reads `id` INSIDE the still-open session -- avoids a DetachedInstanceError below
 
        with engine.begin() as raw_conn:  # => co-11: a SEPARATE raw connection -- bypasses this session's ORM entirely
            raw_conn.execute(text("UPDATE customer SET name = 'Ada Lovelace' WHERE id = :id"), {"id": ada_id})
            # => this UPDATE happens BEHIND the session's back -- `ada.name` in Python memory is now stale
 
        print(f"before refresh: {ada.name}")  # => Output: before refresh: Ada
        # => co-11: the session has NO reason to know the row changed -- it still trusts its own in-memory copy
 
        session.expire(ada)  # => co-11: marks `ada`'s attributes as stale -- does NOT reload them yet, just discards them
        # => expire() alone is lazy: the NEXT attribute access would trigger a reload -- refresh() below makes it eager
        session.refresh(ada)  # => co-11: forces an immediate reload -- issues a fresh SELECT for this object's PK
        print(f"after refresh: {ada.name}")  # => Output: after refresh: Ada Lovelace
        assert ada.name == "Ada Lovelace"  # => co-11: expire() + refresh() is the ORM's explicit "trust the database now"
        # => escape hatch -- without it, an object silently diverges from a row another process (or connection) changed
        # => Example 33 shows the OTHER trigger for a reload: querying, not calling refresh() by hand
    print("ex-30 OK")  # => Output: ex-30 OK

before refresh: Ada
after refresh: Ada Lovelace
ex-30 OK

# pyright: strict
"""Example 31: Unit of Work -- the Session Orders INSERTs by Dependency, Not by Code Order."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, text  # => co-12: event captures the ACTUAL flush order
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-12: the PARENT -- its row must exist before any child's FK can reference it
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => the PK the child FK depends on
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => the parent-to-child collection
 
 
class CustomerOrder(Base):  # => co-12: the CHILD -- its FK depends on the parent's PK already existing
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the dependency the unit of work must respect
    total: Mapped[int] = mapped_column()  # => the order's total in cents, kept as a plain int for this example
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the child-to-parent reverse link
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
 
    statements: list[str] = []  # => every SQL statement the ORM emits, captured for verification below
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        statements.append(" ".join(statement.split()[:3]).upper())  # => records "INSERT INTO <table>" -- verb AND target
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        order = CustomerOrder(total=1999)  # => co-12: the CHILD object is constructed and added FIRST, out of dependency order
        # => deliberately backwards -- proves the unit of work computes dependency order itself, not just source order
        ada = Customer(name="Ada")  # => the PARENT object, constructed SECOND -- code order says child, then parent
        order.customer = ada  # => links them -- back_populates keeps ada.orders in sync too
        session.add(order)  # => registers the child FIRST, matching the CODE order above
        session.add(ada)  # => registers the parent SECOND, again matching the CODE order above
        session.commit()  # => co-12: flushes both -- but NOT in the order they were added
 
    insert_statements = [s for s in statements if s.startswith("INSERT")]  # => filters to the two INSERTs this example expects
    print(insert_statements)  # => Output: ['INSERT INTO CUSTOMER', 'INSERT INTO CUSTOMER_ORDER']
    assert insert_statements == ["INSERT INTO CUSTOMER", "INSERT INTO CUSTOMER_ORDER"]  # => customer landed FIRST
    # => co-12: the unit of work reordered these writes ITSELF -- it inspected the FK dependency (order.customer_id ->
    # => customer.id) and inserted customer BEFORE customer_order, even though `order` was added to the session first
    # => this is the defining trait of Unit of Work: the caller declares WHAT changed, the pattern decides WHEN to write it
    print("ex-31 OK")  # => Output: ex-31 OK

['INSERT INTO CUSTOMER', 'INSERT INTO CUSTOMER_ORDER']
ex-31 OK

# pyright: strict
"""Example 32: Unit of Work -- Dirty Tracking Writes Only Changed Columns."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, create_engine, event, select, text  # => co-12: event captures the emitted UPDATE's own SQL
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-12: three columns, only ONE of which this example mutates
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column -- this one gets mutated below
    email: Mapped[str]  # => a required TEXT column -- this one is NEVER touched after the insert
    country: Mapped[str]  # => a required TEXT column -- this one is NEVER touched after the insert either
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE customer from Customer's Mapped[] fields
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        ada = Customer(name="Ada", email="ada@example.com", country="UK")  # => three columns, all set at insert time
        session.add(ada)  # => registers `ada` as pending
        session.commit()  # => flushes the INSERT, assigns ada.id
        ada_id = ada.id  # => reads `id` INSIDE the still-open session -- avoids a DetachedInstanceError below
 
    statements: list[str] = []  # => captures the RAW SQL of every statement, for inspecting exactly which columns appear
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        statements.append(statement)  # => records the RAW SQL text SQLAlchemy is about to send to Postgres
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- reloads Ada, then mutates ONLY her name
        loaded = session.execute(select(Customer).where(Customer.id == ada_id)).scalar_one()  # => reload by PK
        loaded.name = "Ada Lovelace"  # => co-12: mutating ONE attribute -- the session's dirty-tracking notices THIS one
        # => `loaded.email` and `loaded.country` are untouched -- the unit of work knows the difference between "loaded"
        # => and "changed", because it snapshots every attribute at load time and diffs against that snapshot on flush
        session.commit()  # => flushes ONLY the dirty column(s), then commits
 
    update_statements = [s for s in statements if s.strip().upper().startswith("UPDATE")]  # => filters to the one UPDATE
    print(update_statements)  # => Output: ['UPDATE customer SET name=%(name)s::VARCHAR WHERE customer.id = %(customer_id)s::INTEGER']
    assert len(update_statements) == 1  # => exactly one UPDATE was emitted
    assert "email" not in update_statements[0] and "country" not in update_statements[0]  # => co-12: the UNCHANGED columns
    # => never appear in the SET clause at all -- SQLAlchemy does not blindly re-write every column on every save
    # => contrast with a hand-rolled "UPDATE customer SET name=%s, email=%s, country=%s" -- Active Record libraries that
    # => lack dirty tracking often do exactly that, re-sending every column whether it changed or not
    print("ex-32 OK")  # => Output: ex-32 OK

['UPDATE customer SET name=%(name)s::VARCHAR WHERE customer.id = %(customer_id)s::INTEGER']
ex-32 OK

# pyright: strict
"""Example 33: Unit of Work -- Autoflush Runs the Pending INSERT Before Your SELECT."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import Engine, create_engine, select, text  # => co-12: select() is what TRIGGERS autoflush below
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-12: the mapped class this example queries WHILE another instance is still pending
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE customer from Customer's Mapped[] fields
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table
    with Session(engine) as session:  # => co-12: `autoflush=True` is the Session's DEFAULT -- not set explicitly here
        ada = Customer(name="Ada")  # => a transient object, about to become pending
        session.add(ada)  # => co-12: PENDING -- no INSERT sent yet, no commit() called either
        print(f"autoflush enabled: {session.autoflush}")  # => Output: autoflush enabled: True
 
        rows = session.execute(select(Customer)).scalars().all()  # => co-12: a plain SELECT, no commit() came before it
        # => this is the crux of autoflush: BEFORE running the SELECT against Postgres, the session flushed its own
        # => pending INSERT first -- otherwise the query would miss Ada entirely, contradicting the app's own in-memory
        # => view of the world (ada is clearly "there", the session just hadn't told Postgres about it yet)
        names = [row.name for row in rows]  # => reads the query's own result set
        print(f"names={names}")  # => Output: names=['Ada']
        assert names == ["Ada"]  # => co-12: the pending Ada WAS visible to the query -- autoflush ran first, automatically
        assert ada not in session.new  # => co-12: ada is no longer "new"/pending -- autoflush already promoted it to persistent
 
        session.commit()  # => co-17: commits the transaction the autoflush opened -- makes the write durable
    # => co-12: turning autoflush off (Session(engine, autoflush=False)) is rare and usually a mistake -- queries would
    # => then silently miss your own uncommitted writes, an easy source of "why doesn't my object show up" bugs
    print("ex-33 OK")  # => Output: ex-33 OK

autoflush enabled: True
names=['Ada']
ex-33 OK

# pyright: strict
"""Example 34: Lazy Loading -- Accessing a Relationship Fires an EXTRA SELECT."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, text  # => co-13: event captures the LAZY query firing
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-13: the parent whose relationship this example accesses LAZILY
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy="select"
    # => "select" IS the default -- no lazy=... kwarg above means SQLAlchemy picks this strategy for you
 
 
class CustomerOrder(Base):  # => co-13: the child, only fetched on FIRST access of `.orders`
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK the lazy query filters by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        ada = Customer(name="Ada", orders=[CustomerOrder(total=Decimal("19.99"))])  # => one parent, one child, seeded
        session.add(ada)  # => cascades: registers both the parent and the child as pending
        session.commit()  # => flushes both INSERTs
        ada_id = ada.id  # => reads `id` INSIDE the still-open session -- avoids a DetachedInstanceError below
 
    statements: list[str] = []  # => every SELECT the ORM emits from here on, captured for verification below
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this example only cares about read traffic
            statements.append(" ".join(statement.split()[:2]).upper())  # => records "SELECT ..." -- just enough to count
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the parent fetch below is a real round trip
        fetched = session.get(Customer, ada_id)  # => query #1: fetches ONLY the customer row, no orders yet
        assert fetched is not None  # => narrows Optional[Customer] for pyright --strict below
        print(f"selects after get(): {len(statements)}")  # => Output: selects after get(): 1
        # => co-13: `.orders` has NOT been touched yet -- the parent alone cost exactly one SELECT
 
        totals = [order.total for order in fetched.orders]  # => co-13: FIRST access of `.orders` -- fires the lazy query NOW
        print(f"selects after .orders access: {len(statements)}")  # => Output: selects after .orders access: 2
        assert len(statements) == 2  # => co-13: accessing the relationship added EXACTLY one more SELECT, on demand
        print(totals)  # => Output: [Decimal('19.99')]
        assert totals == [Decimal("19.99")]  # => the lazily-fetched child data is correct, just fetched LATE
        # => co-13: "lazy" means DEFERRED, not "free" -- the query still happens, just at first-touch instead of up front
        # => Example 37 shows the alternative: eager loading pulls this same data during the ORIGINAL query instead
    print("ex-34 OK")  # => Output: ex-34 OK

selects after get(): 1
selects after .orders access: 2
[Decimal('19.99')]
ex-34 OK

# pyright: strict
"""Example 35: Lazy Loading After Session Close -- DetachedInstanceError."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
 
from sqlalchemy import Engine, ForeignKey, create_engine, text  # => co-13: relationship() is what needs the LIVE session
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
from sqlalchemy.orm.exc import DetachedInstanceError  # => co-13/co-11: the exact exception a detached lazy access raises
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-11: the object this example detaches by closing its owning session
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column, safe to read even after detach -- it was already loaded
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: NOT loaded yet, on purpose
 
 
class CustomerOrder(Base):  # => the child, deliberately left un-accessed until AFTER the session closes
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK the lazy query would filter by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    session = Session(engine)  # => opened WITHOUT a `with` block -- this example controls the close() moment explicitly
    ada = Customer(name="Ada", orders=[CustomerOrder(total=Decimal("19.99"))])  # => one parent, one child, seeded
    session.add(ada)  # => cascades: registers both the parent and the child as pending
    session.commit()  # => flushes both INSERTs -- ada is now PERSISTENT, still attached to `session`
 
    print(f"name while attached: {ada.name}")  # => Output: name while attached: Ada
    # => co-13: `.name` is safe -- it was already loaded into `ada.__dict__` by the commit's own refresh, no query needed
 
    session.close()  # => co-11: STAGE 4 -- CLOSE. ada is now DETACHED -- no session backs it anymore
    print(f"name after close: {ada.name}")  # => Output: name after close: Ada
    # => co-11: still safe -- `name` was ALREADY loaded, so reading it needs no round trip to Postgres at all
 
    try:  # => the access below NEEDS a live session, because `.orders` was never touched before close()
        _ = ada.orders  # => co-13: first access of an UNLOADED relationship on a DETACHED object -- no session to query with
        raise AssertionError("expected DetachedInstanceError")  # => co-13: fails loudly if SQLAlchemy's behavior ever changes
    except DetachedInstanceError as exc:  # => co-13 + co-11: the ORM refuses to silently return an empty/wrong list
        print(f"raised: {type(exc).__name__}")  # => Output: raised: DetachedInstanceError
        # => co-13: this is EXACTLY the failure mode eager loading (Example 37) or `AsyncAttrs` (co-16/co-24) sidesteps --
        # => a lazy attribute is a promise to query LATER, and "later" must still be inside a session that is still open
    print("ex-35 OK")  # => Output: ex-35 OK

name while attached: Ada
name after close: Ada
raised: DetachedInstanceError
ex-35 OK

# pyright: strict
"""Example 36: The N+1 Problem -- One Parent Query Fans Out Into N Child Queries."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-15: event captures EVERY query fired
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-15: the N parents this example loops over
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy="select"
 
 
class CustomerOrder(Base):  # => co-15: one child fetched PER PARENT the loop below touches
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK each lazy query filters by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    n_customers = 5  # => co-15: the "N" in "N+1" -- five parents, each with exactly one child order
    # => small ON PURPOSE: even N=5 is enough to show the pattern, and it scales linearly to real production sizes
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n_customers):  # => seeds five customers, one order each
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
            # => cascades: adding the parent also registers its one child, same as Examples 22 and 34
        session.commit()  # => flushes all ten rows (5 customer + 5 customer_order) in one batched transaction
 
    select_statements: list[str] = []  # => every SELECT the ORM emits during the loop below, captured for verification
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this example only cares about read traffic
            select_statements.append(statement)  # => keeps the FULL text -- needed to tell parent SELECTs from child ones
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the count below reflects THIS loop only
        customers = session.execute(select(Customer)).scalars().all()  # => query #1: the ONE parent query
        for customer in customers:  # => co-15: THIS loop is where the N+1 actually happens
            _ = [order.total for order in customer.orders]  # => co-13 + co-15: EACH iteration fires its OWN lazy SELECT
 
    print(f"customers fetched: {len(customers)}")  # => Output: customers fetched: 5
    print(f"total SELECTs: {len(select_statements)}")  # => Output: total SELECTs: 6
    assert len(customers) == n_customers  # => confirms the parent query itself returned all 5 rows correctly
    assert len(select_statements) == n_customers + 1  # => co-15: 1 parent query + 5 child queries = the "N+1" pattern
    # => co-15: the loop LOOKS innocent -- `customer.orders` reads like plain attribute access, not a database call --
    # => but each access is a full round trip to Postgres, and this scales LINEARLY with the number of parents fetched
    # => Example 37 fixes this exact loop with `selectinload()`, dropping the query count from N+1 down to a constant 2
    print("ex-36 OK")  # => Output: ex-36 OK

customers fetched: 5
total SELECTs: 6
ex-36 OK

# pyright: strict
"""Example 37: selectinload() -- Fixing N+1 With a SECOND, BATCHED Query."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-14: event captures the query COUNT
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, selectinload  # => co-14's strategy
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-14: the SAME N parents Example 36 fetched, now loaded eagerly instead
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, OVERRIDDEN below
 
 
class CustomerOrder(Base):  # => co-14: fetched in ONE batched query for ALL parents, not one query PER parent
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK the batched IN-query filters by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    n_customers = 5  # => the same N as Example 36 -- the fix, not the workload, is what changes here
    # => watch the query COUNT below stay flat at 2 no matter how large n_customers grows -- that constancy is the fix
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n_customers):  # => seeds five customers, one order each -- identical setup to Example 36
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
        session.commit()  # => flushes all ten rows in one batched transaction
 
    select_statements: list[str] = []  # => every SELECT the ORM emits during the loop below, captured for verification
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this example only cares about read traffic
            select_statements.append(statement)  # => keeps the FULL text for the IN-clause check below
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the count reflects THIS query alone
        stmt = select(Customer).options(selectinload(Customer.orders))  # => co-14: ONE extra, TARGETED batch-load query
        customers = session.execute(stmt).scalars().all()  # => query #1 (parents) -- selectinload fires query #2 RIGHT HERE
        for customer in customers:  # => co-14: `.orders` is ALREADY loaded -- this loop touches NO new database traffic
            _ = [order.total for order in customer.orders]  # => reads from memory, not a lazy SELECT (contrast Example 36)
 
    print(f"customers fetched: {len(customers)}")  # => Output: customers fetched: 5
    print(f"total SELECTs: {len(select_statements)}")  # => Output: total SELECTs: 2
    assert len(select_statements) == 2  # => co-14 + co-15: ONE query for parents, ONE batched query for ALL children
    assert "customer_order.customer_id IN" in select_statements[1]  # => co-14: the SECOND query uses a single IN (...) list
    # => this IN-clause is the batching mechanism itself -- it names every parent PK collected from query #1, at once
    # => co-14: selectinload() issues `SELECT ... FROM customer_order WHERE customer_id IN (1, 2, 3, 4, 5)` -- ONE round
    # => trip covers every parent's children at once, instead of Example 36's five separate per-parent SELECTs
    print("ex-37 OK")  # => Output: ex-37 OK

customers fetched: 5
total SELECTs: 2
ex-37 OK

# pyright: strict
"""Example 38: joinedload() -- Fixing N+1 With ONE Single JOIN Query."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-14: event captures the query COUNT
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, joinedload, mapped_column, relationship  # => co-14's strategy
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-14: the SAME N parents Examples 36-37 fetched, now via a single JOIN
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, OVERRIDDEN below
 
 
class CustomerOrder(Base):  # => co-14: fetched IN THE SAME ROW SET as its parent -- no second query at all
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK the JOIN's ON clause matches on
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    n_customers = 5  # => the same N as Examples 36-37 -- only the loading STRATEGY changes here
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n_customers):  # => seeds five customers, one order each -- identical setup to Examples 36-37
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
        session.commit()  # => flushes all ten rows in one batched transaction
 
    select_statements: list[str] = []  # => every SELECT the ORM emits during the query below, captured for verification
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this example only cares about read traffic
            select_statements.append(statement)  # => keeps the FULL text for the JOIN check below
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the count reflects THIS query alone
        stmt = select(Customer).options(joinedload(Customer.orders))  # => co-14: an OUTER JOIN pulls both tables at once
        customers = session.execute(stmt).unique().scalars().all()  # => .unique() dedupes parents repeated by the JOIN
        # => co-14: without .unique(), a customer with 2+ orders would appear TWICE -- one row per JOINed match
        for customer in customers:  # => co-14: `.orders` is ALREADY loaded -- this loop touches NO new database traffic
            _ = [order.total for order in customer.orders]  # => reads from memory, populated by the single JOIN above
 
    print(f"customers fetched: {len(customers)}")  # => Output: customers fetched: 5
    print(f"total SELECTs: {len(select_statements)}")  # => Output: total SELECTs: 1
    assert len(select_statements) == 1  # => co-14 + co-15: ONE query total -- parents AND children in the SAME result set
    assert "LEFT OUTER JOIN customer_order" in select_statements[0]  # => co-14: the ONE query IS a JOIN, not two SELECTs
    # => co-14: joinedload() trades query COUNT for row-set SIZE -- one round trip, but a wider, sometimes-duplicated
    # => result set that needs .unique() to dedupe; selectinload() (Example 37) trades the other way: two queries, but
    # => each one stays narrow -- which strategy wins depends on how many children each parent typically has (co-14)
    print("ex-38 OK")  # => Output: ex-38 OK

customers fetched: 5
total SELECTs: 1
ex-38 OK

# pyright: strict
"""Example 39: subqueryload() -- a LEGACY Batched Strategy, Superseded by selectinload()."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-14: event captures the query COUNT
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, subqueryload  # => co-14's LEGACY strategy
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-14: the SAME N parents Examples 36-38 fetched, now via subqueryload
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, OVERRIDDEN below
 
 
class CustomerOrder(Base):  # => co-14: fetched via a SEPARATE query that RE-RUNS the original SELECT as a subquery
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK the batched subquery filters by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    n_customers = 5  # => the same N as Examples 36-38 -- only the loading STRATEGY changes here
    # => keeping the workload identical across Examples 36-39 is what makes the query-count contrast in Example 40 fair
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n_customers):  # => seeds five customers, one order each -- identical setup to Examples 36-38
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
        session.commit()  # => flushes all ten rows in one batched transaction
 
    select_statements: list[str] = []  # => every SELECT the ORM emits during the query below, captured for verification
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this example only cares about read traffic
            select_statements.append(statement)  # => keeps the FULL text for the subquery check below
 
    event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches the hook to every statement on `engine`
 
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the count reflects THIS query alone
        stmt = select(Customer).options(subqueryload(Customer.orders))  # => co-14: LEGACY -- re-embeds the ORIGINAL query
        customers = session.execute(stmt).scalars().all()  # => query #1 (parents); subqueryload fires query #2 RIGHT HERE
        for customer in customers:  # => co-14: `.orders` is ALREADY loaded -- this loop touches NO new database traffic
            _ = [order.total for order in customer.orders]  # => reads from memory, populated by the second query above
 
    print(f"customers fetched: {len(customers)}")  # => Output: customers fetched: 5
    print(f"total SELECTs: {len(select_statements)}")  # => Output: total SELECTs: 2
    assert len(select_statements) == 2  # => co-14: also 2 queries, like selectinload -- the difference is HOW #2 filters
    assert "FROM (SELECT customer.id AS customer_id" in select_statements[1]  # => co-14: an embedded subquery re-derives the parents
    # => co-14: SQLAlchemy's own docs now call subqueryload() "mostly legacy" -- selectinload() (Example 37) does the
    # => SAME job with a simpler, more predictable IN (...) list instead of re-running the parent query as a correlated
    # => subquery; prefer selectinload() in new code -- subqueryload() mainly still matters for reading OLDER codebases
    print("ex-39 OK")  # => Output: ex-39 OK

customers fetched: 5
total SELECTs: 2
ex-39 OK

# pyright: strict
"""Example 40: Eager-Loading Strategy Contrast -- selectinload vs joinedload vs subqueryload."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => the event hook's callback signature is untyped by SQLAlchemy's own stubs
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-14: event captures each STRATEGY's shape
from sqlalchemy.orm import (  # => co-14: all three loading strategies compared side by side
    DeclarativeBase,  # => the shared mapper registry base for Customer/CustomerOrder below
    Mapped,  # => typed column annotation (DD-39) -- pyright --strict enforces every field
    Session,  # => the unit-of-work handle each strategy's query runs through (co-12)
    joinedload,  # => strategy #2: a single-query LEFT OUTER JOIN
    mapped_column,  # => the runtime column constructor Mapped[] pairs with
    relationship,  # => declares the one-to-many Customer.orders collection
    selectinload,  # => strategy #1: a batched, second IN-clause query
    subqueryload,  # => strategy #3: a legacy, embedded-subquery second query
)
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-14: the SAME schema Examples 36-39 used -- only the OPTIONS below change per run
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, OVERRIDDEN per run
 
 
class CustomerOrder(Base):  # => the child whose loading strategy this example varies across three runs
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK every strategy below joins/filters on
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_and_seed(engine: Engine, n_customers: int) -> None:  # => shared setup -- fresh schema, N customers, one order each
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fresh state for each strategy's run
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n_customers):  # => seeds N customers, one order each -- IDENTICAL workload for every strategy
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
        session.commit()  # => flushes all rows before this strategy's own query count starts being measured
 
 
def count_selects(engine: Engine, option: Any) -> int:  # => co-14: `option` is a loader-option object -- SQLAlchemy's stubs
    # => leave its precise type broad, so this ONE helper works for selectinload/joinedload/subqueryload interchangeably
    statements: list[str] = []  # => every SELECT this ONE strategy's query fires, reset per call
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this helper only cares about read traffic
            statements.append(statement)  # => records every SELECT text for this strategy's single query round
 
    listener = event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches for the duration of this call
    with Session(engine) as session:  # => a FRESH session -- nothing cached, so the count reflects THIS strategy alone
        stmt = select(Customer).options(option)  # => co-14: the ONLY thing that differs between the three calls below
        customers = session.execute(stmt).unique().scalars().all()  # => .unique() is a harmless no-op for the non-JOIN strategies
        for customer in customers:  # => touches every child so a LAZY fallback would also show up in the count
            _ = [order.total for order in customer.orders]  # => reads from memory if eager-loaded, or fires a lazy SELECT if not
    event.remove(engine, "before_cursor_execute", listener)  # => detaches so the NEXT strategy's call starts from zero
    return len(statements)  # => the total SELECT count this ONE strategy needed for the whole workload
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine, reused across all three strategy runs
    n_customers = 5  # => the same N used in Examples 36-39, so the counts below are directly comparable
    reset_and_seed(engine, n_customers)  # => fresh, identical workload before the FIRST strategy's measurement
    selectin_count = count_selects(engine, selectinload(Customer.orders))  # => co-14: batched IN-clause strategy
 
    reset_and_seed(engine, n_customers)  # => re-seed -- each strategy measures against its OWN clean baseline
    joined_count = count_selects(engine, joinedload(Customer.orders))  # => co-14: single-query JOIN strategy
 
    reset_and_seed(engine, n_customers)  # => re-seed once more for the third and final strategy
    subquery_count = count_selects(engine, subqueryload(Customer.orders))  # => co-14: legacy embedded-subquery strategy
 
    print(f"selectinload: {selectin_count} queries")  # => Output: selectinload: 2 queries
    print(f"joinedload: {joined_count} queries")  # => Output: joinedload: 1 queries
    print(f"subqueryload: {subquery_count} queries")  # => Output: subqueryload: 2 queries
    assert (selectin_count, joined_count, subquery_count) == (2, 1, 2)  # => co-14: joinedload wins on COUNT, always 1
    # => co-14: joinedload's single query can still be the SLOWER choice once a parent has many children -- the JOIN's
    # => row set repeats every parent column once per child row, while selectinload/subqueryload keep two narrow,
    # => non-duplicated result sets; prefer selectinload() as the default modern choice, joinedload() for TRUE 1:1s,
    # => and treat subqueryload() as legacy -- present here only to recognize it when reading an older codebase
    print("ex-40 OK")  # => Output: ex-40 OK

selectinload: 2 queries
joinedload: 1 queries
subqueryload: 2 queries
ex-40 OK

# pyright: strict
"""Example 41: raiseload() -- Turning an Accidental Lazy Load Into a Loud Error."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
 
from sqlalchemy import Engine, ForeignKey, create_engine, select, text  # => co-16: select() is the query raiseload guards
from sqlalchemy.exc import InvalidRequestError  # => co-16: the exact exception a guarded lazy access raises
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, raiseload, relationship  # => co-16's guard
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-16: the parent this example queries WITH the raiseload guard applied
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, GUARDED below
 
 
class CustomerOrder(Base):  # => co-16: the relationship this example forbids touching by accident
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK a lazy query would have filtered by
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer and customer_order tables
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        ada = Customer(name="Ada", orders=[CustomerOrder(total=Decimal("19.99"))])  # => one parent, one child, seeded
        session.add(ada)  # => cascades: registers both the parent and the child as pending
        session.commit()  # => flushes both INSERTs
        ada_id = ada.id  # => reads `id` INSIDE the still-open session -- avoids a DetachedInstanceError below
 
    with Session(engine) as session:  # => a FRESH session -- the guard below applies to THIS query only
        stmt = select(Customer).options(raiseload(Customer.orders))  # => co-16: forbids `.orders` from ever lazy-loading
        fetched = session.execute(stmt).scalar_one()  # => fetches ONLY the customer row -- exactly like the default
        print(f"name: {fetched.name}")  # => Output: name: Ada
        # => co-16: fields already covered by the SELECT (id, name) are still perfectly readable -- only the RELATIONSHIP is guarded
 
        try:  # => the access below is EXACTLY what Example 36's N+1 loop did by accident, forbidden here on purpose
            _ = fetched.orders  # => co-16: touching a raiseload()-guarded relationship never silently queries
            raise AssertionError("expected InvalidRequestError")  # => fails loudly if SQLAlchemy's behavior ever changes
        except InvalidRequestError as exc:  # => co-16: a LOUD, immediate error instead of a silent extra round trip
            print(f"raised: {type(exc).__name__}")  # => Output: raised: InvalidRequestError
            # => co-16: raiseload() converts "forgot to eager-load" from a silent performance bug into a crash you catch
            # => in code review or a test run -- the SAME safety idea AsyncAttrs (co-24) applies for async sessions, where
            # => an accidental lazy load is not just slow but structurally FORBIDDEN by the driver itself (see Example 62)
    print("ex-41 OK")  # => Output: ex-41 OK

name: Ada
raised: InvalidRequestError
ex-41 OK

# pyright: strict
"""Example 42: Asserting the Query Count -- Before and After an N+1 Fix, in One Script."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
from collections.abc import Generator  # => the modern return-type annotation @contextmanager expects, not Iterator
from contextlib import contextmanager  # => co-15: a reusable "count queries in this block" helper
from decimal import Decimal  # => money is Decimal, never float -- exact cents, no rounding drift
from typing import Any  # => types SQLAlchemy's own untyped event-hook callback arguments
 
from sqlalchemy import Engine, ForeignKey, create_engine, event, select, text  # => co-15: event is the counting mechanism itself
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, selectinload  # => co-14's fix
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-15: the parents this example's assertion counts queries against
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
    orders: Mapped[list["CustomerOrder"]] = relationship(back_populates="customer")  # => co-13: default lazy, OVERRIDDEN below
 
 
class CustomerOrder(Base):  # => the child whose access pattern this example counts, before and after the fix
    __tablename__ = "customer_order"  # => named to avoid the reserved SQL word "order"
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))  # => the FK every counted query touches
    total: Mapped[Decimal]  # => the order's total, as an exact Decimal
    customer: Mapped[Customer] = relationship(back_populates="orders")  # => the reverse, many-to-one navigation
 
 
def reset_and_seed(engine: Engine, n: int) -> None:  # => shared setup -- fresh schema, N customers, one order each
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fresh state for each measurement
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build both tables into
    Base.metadata.create_all(engine)  # => issues CREATE TABLE for both customer and customer_order
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        for i in range(n):  # => seeds N customers, one order each
            session.add(Customer(name=f"Customer{i}", orders=[CustomerOrder(total=Decimal("9.99"))]))
        session.commit()  # => flushes all rows before this measurement's own counter starts
 
 
@contextmanager  # => co-15: turns "count every SELECT fired inside this block" into a plain `with` statement
def query_counter(engine: Engine) -> Generator[list[int]]:  # => yields a one-element mutable box holding the running count
    box = [0]  # => a list, not a plain int -- the caller reads box[0] AFTER the block exits, still seeing live updates
 
    def on_execute(conn: Any, cursor: Any, statement: str, *rest: Any) -> None:  # => untyped hook params (SQLAlchemy's own)
        if statement.strip().upper().startswith("SELECT"):  # => this counter only cares about read traffic
            box[0] += 1  # => increments the SAME box the caller holds a reference to
 
    listener = event.listens_for(engine, "before_cursor_execute")(on_execute)  # => attaches for the block's duration
    try:  # => the caller's code runs HERE, between attach and detach
        yield box  # => hands the box to the `with` block -- readable both during and after
    finally:  # => detaches even if the caller's block raises
        event.remove(engine, "before_cursor_execute", listener)  # => co-15: cleanup -- the NEXT measurement starts at zero
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine, reused across BEFORE and AFTER measurements
    n_customers = 5  # => the same N as Examples 36-40, so this assertion is directly comparable to those results
    # => a REAL regression test would parameterize this and assert `after_count == 2` regardless of N, catching drift
 
    reset_and_seed(engine, n_customers)  # => fresh workload for the BEFORE measurement
    with query_counter(engine) as before_count:  # => co-15: measures the UNFIXED, lazy-default access pattern
        with Session(engine) as session:  # => a FRESH session -- nothing cached
            customers = session.execute(select(Customer)).scalars().all()  # => query #1, the one parent query
            for customer in customers:  # => this loop is the N+1 itself -- ONE lazy SELECT per parent
                _ = [order.total for order in customer.orders]  # => co-13: each access fires its own round trip
    print(f"before fix: {before_count[0]} queries")  # => Output: before fix: 6 queries
    assert before_count[0] == n_customers + 1  # => co-15: asserts the UNFIXED count is exactly N+1, not an estimate
 
    reset_and_seed(engine, n_customers)  # => fresh workload for the AFTER measurement -- same shape, same size
    with query_counter(engine) as after_count:  # => co-15: measures the SAME access pattern, now eager-loaded
        with Session(engine) as session:  # => a FRESH session -- nothing cached
            stmt = select(Customer).options(selectinload(Customer.orders))  # => co-14: the fix from Example 37
            customers = session.execute(stmt).scalars().all()  # => query #1 (parents) + query #2 (batched children)
            for customer in customers:  # => identical loop, but `.orders` is now already loaded -- no new traffic
                _ = [order.total for order in customer.orders]  # => reads from memory, contributes zero new SELECTs
    print(f"after fix: {after_count[0]} queries")  # => Output: after fix: 2 queries
    assert after_count[0] == 2  # => co-15: the fix collapses N+1 down to a CONSTANT 2, regardless of how large N grows
    # => try raising n_customers to 50 or 500 and re-running -- before_count grows linearly, after_count stays at 2
    # => co-15: this is the pattern a CI regression test should assert directly -- a hard-coded query-count ceiling
    # => catches a reintroduced N+1 the moment someone accidentally removes an eager-load option, before it ships
    print("ex-42 OK")  # => Output: ex-42 OK

before fix: 6 queries
after fix: 2 queries
ex-42 OK

# pyright: strict
"""Example 43: session.begin() -- Commit on Success, Rollback on Error."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import Engine, create_engine, select, text  # => co-17: select() confirms what actually survived
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-17: the table this example writes to, once successfully and once NOT
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column, UNIQUE below to force the second attempt to fail
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
        conn.execute(text("CREATE TABLE customer (id SERIAL PRIMARY KEY, name TEXT UNIQUE NOT NULL)"))  # => raw DDL, adds UNIQUE
    # => Base.metadata.create_all() is skipped here on purpose -- the UNIQUE constraint above needs raw DDL, since
    # => Customer's own Mapped[str] doesn't declare uniqueness, and this example wants a REAL constraint violation
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table with a UNIQUE name constraint
 
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        with session.begin():  # => co-17: an EXPLICIT transaction scope -- commits automatically on a clean exit
            session.add(Customer(name="Ada"))  # => the FIRST, successful write
        # => co-17: `session.begin()`'s own `with` block already committed here -- no separate session.commit() needed
 
        try:  # => the block below deliberately violates the UNIQUE constraint on `name`
            with session.begin():  # => co-17: a SECOND transaction scope -- rolls back automatically on any exception
                session.add(Customer(name="Ada"))  # => the SAME name -- Postgres will reject this at flush/commit time
        except Exception as exc:  # => co-17: the UNIQUE violation propagates as a real exception, not a silent no-op
            print(f"raised: {type(exc).__name__}")  # => Output: raised: IntegrityError
            # => co-17: `session.begin()`'s `with` block caught the exception, rolled back, and RE-RAISED it here --
            # => the failed second Ada was never durably written, even though `session.add()` itself never complained
 
    with Session(engine) as session:  # => a FRESH session, just to read back the final state
        names = session.execute(select(Customer.name)).scalars().all()  # => co-17: confirms what actually persisted
        print(f"names={names}")  # => Output: names=['Ada']
        assert names == ["Ada"]  # => co-17: exactly ONE Ada survived -- the committed write, not the rolled-back one
    # => co-17: `with session.begin():` is the idiomatic pattern -- commit on success, rollback on ANY exception,
    # => with no manual try/except/commit/rollback bookkeeping required at the call site itself
    print("ex-43 OK")  # => Output: ex-43 OK

raised: IntegrityError
names=['Ada']
ex-43 OK

# pyright: strict
"""Example 44: begin_nested() -- a Savepoint That Rolls Back PART of a Transaction."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import Engine, create_engine, select, text  # => co-17: select() confirms exactly which rows survived
from sqlalchemy.exc import IntegrityError  # => co-17: the exact exception the inner savepoint's violation raises
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Customer(Base):  # => co-17: the table this example writes to across an outer transaction and an inner savepoint
    __tablename__ = "customer"  # => the physical table name
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column, UNIQUE below to force the savepoint's own write to fail
 
 
def reset_schema(engine: Engine) -> None:  # => shared reset helper -- wipes the whole schema, self-contained
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema to build Customer's table into
        conn.execute(text("CREATE TABLE customer (id SERIAL PRIMARY KEY, name TEXT UNIQUE NOT NULL)"))  # => raw DDL, adds UNIQUE
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine
    reset_schema(engine)  # => fresh, empty customer table with a UNIQUE name constraint
 
    with Session(engine) as session:  # => a Session is the ORM's unit-of-work handle (co-12)
        with session.begin():  # => co-17: the OUTER transaction -- this one commits at the very end, wrapping everything
            session.add(Customer(name="Ada"))  # => write #1, part of the OUTER transaction, survives regardless
            session.add(Customer(name="Grace"))  # => write #2, ALSO part of the OUTER transaction, survives regardless
 
            try:  # => an INNER savepoint, nested inside the still-open outer transaction
                with session.begin_nested():  # => co-17: begin_nested() opens a Postgres SAVEPOINT, not a whole new transaction
                    session.add(Customer(name="Ada"))  # => a DUPLICATE name -- violates UNIQUE, but only WITHIN this savepoint
            except IntegrityError:  # => co-17: begin_nested()'s `with` block rolls back to the SAVEPOINT, not the whole transaction
                print("inner savepoint rolled back")  # => Output: inner savepoint rolled back
                # => co-17: Ada and Grace from OUTSIDE the savepoint are untouched -- only the failed duplicate write is undone
 
    with Session(engine) as session:  # => a FRESH session, just to read back the final state
        names = sorted(session.execute(select(Customer.name)).scalars().all())  # => co-17: confirms exactly what survived
        print(f"names={names}")  # => Output: names=['Ada', 'Grace']
        assert names == ["Ada", "Grace"]  # => co-17: BOTH outer writes committed -- the inner savepoint's failure was ISOLATED
    # => co-17: this is the key difference from Example 43 -- there, the WHOLE transaction rolled back on failure;
    # => here, begin_nested() lets a caller attempt a risky write, catch its failure, and keep the surrounding
    # => transaction alive -- useful for "try this insert, fall back to something else" without losing prior work
    print("ex-44 OK")  # => Output: ex-44 OK

inner savepoint rolled back
names=['Ada', 'Grace']
ex-44 OK

# pyright: strict
"""Example 45: pool_size and max_overflow -- Connections Reused, Not Reopened Per Query."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import create_engine, text  # => co-18: the engine owns the pool every connection borrows from
from sqlalchemy.pool import QueuePool  # => co-18: the default pool implementation this example inspects directly
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL, pool_size=2, max_overflow=1)  # => co-18: at most 2 PERSISTENT + 1 overflow connection
    pool = engine.pool  # => co-18: the actual QueuePool object -- inspectable at runtime, not just a config knob
    assert isinstance(pool, QueuePool)  # => confirms the DEFAULT pool class -- narrows the type for pyright --strict below
 
    print(f"checked out before use: {pool.checkedout()}")  # => Output: checked out before use: 0
    # => co-18: create_engine() does NOT open any connections up front -- the pool starts completely empty
 
    with engine.connect() as conn1:  # => co-18: borrows connection #1 FROM the pool -- opens a NEW physical connection
        pid1 = conn1.execute(text("SELECT pg_backend_pid()")).scalar_one()  # => Postgres' own process id for THIS connection
        print(f"checked out with 1 open: {pool.checkedout()}")  # => Output: checked out with 1 open: 1
 
        with engine.connect() as conn2:  # => borrows connection #2 -- ALSO a new physical connection, still within pool_size
            pid2 = conn2.execute(text("SELECT pg_backend_pid()")).scalar_one()  # => a DIFFERENT backend pid than conn1's
            print(f"checked out with 2 open: {pool.checkedout()}")  # => Output: checked out with 2 open: 2
            assert pid1 != pid2  # => co-18: two SIMULTANEOUS connections are genuinely two separate Postgres backends
 
    print(f"checked out after both closed: {pool.checkedout()}")  # => Output: checked out after both closed: 0
    # => co-18: closing a connection RETURNS it to the pool -- the underlying TCP socket to Postgres often stays open
 
    with engine.connect() as conn3:  # => co-18: borrows a connection AGAIN, after both prior ones were returned
        pid3 = conn3.execute(text("SELECT pg_backend_pid()")).scalar_one()  # => this connection's backend pid
    print(f"pid3 reused an existing pid: {pid3 in (pid1, pid2)}")  # => Output: pid3 reused an existing pid: True
    assert pid3 in (pid1, pid2)  # => co-18: the pool handed back an EXISTING physical connection, not a freshly-opened one
    # => co-18: which of the two it picked depends on FIFO queue order, an implementation detail -- what matters is
    # => that NO third Postgres backend process was ever spawned; opening a NEW connection costs a TCP handshake
    # => plus authentication on every request, so a pool amortizes that cost across many borrow/return cycles instead
    engine.dispose()  # => co-18: closes every pooled connection -- good hygiene at process shutdown, used here for cleanliness
    print("ex-45 OK")  # => Output: ex-45 OK

checked out before use: 0
checked out with 1 open: 1
checked out with 2 open: 2
checked out after both closed: 0
pid3 reused an existing pid: True
ex-45 OK

# pyright: strict
"""Example 46: Pool Exhaustion -- QueuePool Times Out When Every Connection Is Held."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
from sqlalchemy import create_engine, text  # => co-18: the engine whose small pool this example deliberately exhausts
from sqlalchemy.exc import TimeoutError as SATimeoutError  # => co-18: renamed to avoid shadowing the builtin TimeoutError
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(  # => co-18: a DELIBERATELY tiny pool -- easy to exhaust for this demonstration
        SQLA_URL,
        pool_size=1,  # => at most ONE persistent connection
        max_overflow=0,  # => co-18: zero overflow -- no temporary extra connections allowed beyond pool_size
        pool_timeout=1,  # => co-18: how long a checkout WAITS for a free connection before giving up, in seconds
    )
 
    held_conn = engine.connect()  # => co-18: checks out the pool's ONLY connection and does NOT return it -- held open
    # => this simulates a real hazard: a request that forgets to close its connection, or one that runs unusually long
    try:  # => the checkout below has nowhere to come from -- pool_size=1 is already fully checked out
        try:  # => attempts a SECOND checkout while the first is still held
            engine.connect()  # => co-18: blocks up to pool_timeout seconds, then raises -- there is no connection to give
            raise AssertionError("expected TimeoutError")  # => fails loudly if the pool unexpectedly had room
        except SATimeoutError as exc:  # => co-18: the exact exception QueuePool raises once pool_timeout elapses
            print(f"raised: {type(exc).__name__}")  # => Output: raised: TimeoutError
            # => co-18: the message names QueuePool explicitly -- "QueuePool limit of size 1 overflow 0 reached"
    finally:  # => always releases the held connection, even if the assertion above somehow fired
        held_conn.close()  # => co-18: returns the connection to the pool -- the NEXT checkout would now succeed instantly
 
    with engine.connect() as conn_after:  # => co-18: proves the pool RECOVERS once a connection is returned
        result = conn_after.execute(text("SELECT 1")).scalar_one()  # => a trivial round trip, just to prove the pool works
        print(f"result after recovery: {result}")  # => Output: result after recovery: 1
        assert result == 1  # => co-18: exhaustion is TEMPORARY -- the pool is healthy again as soon as a slot frees up
    # => co-18: pool_timeout is a deliberate design choice, not a bug -- it turns "every connection is busy" into a
    # => bounded, catchable wait instead of an indefinitely hanging request; a real service sizes pool_size + max_overflow
    # => to its expected concurrent load, and treats a TimeoutError here as a signal to scale the pool or the database
    engine.dispose()  # => closes every pooled connection -- good hygiene at process shutdown, used here for cleanliness
    print("ex-46 OK")  # => Output: ex-46 OK

raised: TimeoutError
result after recovery: 1
ex-46 OK

# pyright: strict
"""Example 47: pool_pre_ping -- Transparently Recovering a Stale Pooled Connection."""
 
from __future__ import annotations
 
import os  # => reads connection settings from the environment
 
import psycopg  # => co-18: a RAW driver connection for the killer -- must be a DIFFERENT physical backend than the pool's
from sqlalchemy import Engine, create_engine, text  # => co-18: pool_pre_ping is a create_engine() keyword, not a pool method
from sqlalchemy.exc import OperationalError  # => co-18: what a STALE connection raises WITHOUT pre-ping's protection
 
PG_DSN: str = os.environ.get("PG_DSN", "postgresql://postgres:postgres@localhost:5432/orm_by_example")  # => a plain DB-API DSN
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
def make_stale(engine: Engine) -> None:  # => borrows the pool's connection, notes its pid, then kills it FROM OUTSIDE the pool
    with engine.connect() as conn:  # => co-18: checks out the pool's connection -- this becomes the STALE one once returned
        pid = conn.execute(text("SELECT pg_backend_pid()")).scalar_one()  # => Postgres' own process id for THIS connection
    with psycopg.connect(PG_DSN, autocommit=True) as killer:  # => a connection OUTSIDE the pool -- can't kill itself by accident
        killer.execute("SELECT pg_terminate_backend(%s)", (pid,))  # => forcibly kills the pool's connection, not this one
    # => co-18: the pool does NOT know yet -- `conn`'s connection object is back in the pool, looking perfectly healthy
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    no_ping_engine = create_engine(SQLA_URL, pool_size=1, max_overflow=0)  # => co-18: pool_pre_ping DEFAULTS to False
    make_stale(no_ping_engine)  # => kills the pool's one connection behind its back
    try:  # => the pool hands back the connection it THINKS is fine -- it never verified
        with no_ping_engine.connect() as conn:  # => co-18: reuses the now-DEAD connection, no health check first
            conn.execute(text("SELECT 1"))  # => this round trip fails -- Postgres already closed the socket
        raise AssertionError("expected OperationalError")  # => fails loudly if the connection was somehow still alive
    except OperationalError as exc:  # => co-18: without pre-ping, staleness surfaces as a hard failure on the CALLER's query
        print(f"no pre_ping raised: {type(exc).__name__}")  # => Output: no pre_ping raised: OperationalError
    no_ping_engine.dispose()  # => closes every pooled connection -- good hygiene before building the next engine
 
    ping_engine = create_engine(SQLA_URL, pool_size=1, max_overflow=0, pool_pre_ping=True)  # => co-18: the protection ENABLED
    make_stale(ping_engine)  # => kills THIS pool's one connection behind its back, identically to the run above
    with ping_engine.connect() as conn:  # => co-18: pre_ping runs a lightweight "SELECT 1" BEFORE handing this out
        result = conn.execute(text("SELECT 1")).scalar_one()  # => co-18: succeeds -- pre-ping silently opened a FRESH connection
        print(f"with pre_ping result: {result}")  # => Output: with pre_ping result: 1
        assert result == 1  # => co-18: the caller never even sees the staleness -- pre-ping absorbed it transparently
    # => co-18: pool_pre_ping adds one extra round trip PER checkout, a small, worthwhile cost for connections that
    # => might have gone stale behind a load balancer, a database failover, or a long idle period -- exactly the
    # => class of production incident ("connection reset" errors after a quiet night) this setting quietly prevents
    ping_engine.dispose()  # => closes every pooled connection -- good hygiene at process shutdown, used here for cleanliness
    print("ex-47 OK")  # => Output: ex-47 OK

no pre_ping raised: OperationalError
with pre_ping result: 1
ex-47 OK

# pyright: strict
"""Example 48: alembic init -- Scaffolding a Migration Environment."""
 
from __future__ import annotations
 
import contextlib  # => co-19: redirects alembic's own console chatter, which embeds a NON-DETERMINISTIC temp path
import io  # => the throwaway buffer contextlib.redirect_stdout writes alembic's chatter INTO, instead of the real terminal
import os  # => reads connection settings, and builds/removes the scratch project directory
import shutil  # => co-19: cleanup -- removes the scaffolded project directory once this example is done
import tempfile  # => co-19: a fresh, self-contained directory for the scaffolded migration project
 
from alembic import command  # => co-19: the SAME Python API `alembic` init/revision/upgrade CLI commands call internally
from alembic.config import Config  # => co-19: a Config object is what EVERY alembic.command function needs as its first arg
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    project_dir = tempfile.mkdtemp(prefix="alembic_ex48_")  # => co-19: a throwaway directory -- this example's own "repo root"
    script_dir = os.path.join(project_dir, "migrations")  # => co-19: alembic's convention -- a "migrations" subfolder
    ini_path = os.path.join(project_dir, "alembic.ini")  # => co-19: alembic.ini -- the CLI's own config file, generated below
 
    cfg = Config(ini_path)  # => co-19: this Config object IS what `alembic init` would populate from a real alembic.ini
    cfg.set_main_option("script_location", script_dir)  # => co-19: tells Config where the scaffolded files should land
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Creating directory /tmp/xyz... done" -- path varies per run
        command.init(cfg, script_dir)  # => co-19: THE scaffolding step -- identical to running `alembic init migrations`
 
    created_files = sorted(os.listdir(script_dir))  # => co-19: what `alembic init` actually wrote to disk, read back fresh
    print(f"created files: {created_files}")  # => Output: created files: ['README', 'env.py', 'script.py.mako', 'versions']
    assert created_files == ["README", "env.py", "script.py.mako", "versions"]  # => co-19: the exact scaffold, every time
    # => co-19: env.py wires alembic to a real database connection and your models; script.py.mako is the TEMPLATE every
    # => new revision file is generated from; versions/ is where individual migration scripts accumulate over time
 
    versions_dir = os.path.join(script_dir, "versions")  # => co-19: the directory EVERY future migration script lands in
    print(f"versions is a directory: {os.path.isdir(versions_dir)}")  # => Output: versions is a directory: True
    print(f"versions starts empty: {os.listdir(versions_dir) == []}")  # => Output: versions starts empty: True
    assert os.path.isdir(versions_dir) and os.listdir(versions_dir) == []  # => co-19: a fresh project has NO migrations yet
    # => co-19: `alembic init` only scaffolds the ENVIRONMENT -- it writes no migration of its own; Example 49 writes
    # => the FIRST real migration into this exact versions/ directory, using the same Config + command API shown here
 
    shutil.rmtree(project_dir)  # => co-19: cleanup -- this example's own throwaway project, not a real repository
    print("ex-48 OK")  # => Output: ex-48 OK

created files: ['README', 'env.py', 'script.py.mako', 'versions']
versions is a directory: True
versions starts empty: True
ex-48 OK

# pyright: strict
"""Example 49: A Hand-Written create-table Migration -- upgrade() Actually Creates the Table."""
 
from __future__ import annotations
 
import contextlib  # => co-19: swallows alembic's own non-deterministic scaffolding chatter, same reason as Example 48
import io  # => the throwaway buffer redirect_stdout writes that chatter INTO
import os  # => reads connection settings, and builds/removes the scratch project directory
import shutil  # => co-19: cleanup -- removes the scaffolded project directory once this example is done
import tempfile  # => co-19: a fresh, self-contained directory for the scaffolded migration project
 
from alembic import command  # => co-19: init/revision/upgrade -- the same API the alembic CLI itself calls
from alembic.config import Config  # => co-19: every alembic.command function needs one of these as its first argument
from sqlalchemy import create_engine, inspect, text  # => co-19: inspect() reads back what the migration ACTUALLY built
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
# => co-19: a MINIMAL, deterministic env.py this example WRITES to disk below -- no fileConfig(), no logging noise,
# => just the online-migration path; real projects commonly hand-edit env.py exactly like this, right after
# => `alembic init` scaffolds a generic template, to wire it to their own database URL and model metadata
ENV_PY = (  # => co-19: built as SEPARATE joined string lines, not one big block, so each line can carry its own comment
    "from alembic import context\n"  # => co-19: `context` is alembic's OWN migration-runtime handle, not SQLAlchemy's
    "from sqlalchemy import engine_from_config, pool\n\n"  # => co-19: builds a real Engine straight from alembic.ini's own section
    "config = context.config\n"  # => co-19: the SAME Config object bootstrap() constructed, now read back inside env.py
    'target_metadata = config.attributes.get("target_metadata")\n\n\n'  # => None here; Examples 51-52 set this for autogenerate
    "def run_migrations_online() -> None:\n"  # => co-19: THE function alembic's runtime calls for a live-connection migration
    "    connectable = engine_from_config(\n"  # => co-19: reads the sqlalchemy.* keys straight out of alembic.ini's own section
    '        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool\n'  # => co-19: no pool reuse needed
    "    )\n"  # => a short-lived migration run has no reason to KEEP a pooled connection open afterward
    "    with connectable.connect() as connection:\n"  # => borrows ONE connection for the whole migration run
    "        context.configure(connection=connection, target_metadata=target_metadata)\n"  # => co-19: binds THIS run's connection
    "        with context.begin_transaction():\n"  # => co-19: wraps every upgrade()/downgrade() call in ONE transaction
    "            context.run_migrations()\n\n\n"  # => co-19: the actual dispatch -- calls upgrade()/downgrade() on each revision
    "run_migrations_online()\n"  # => co-19: executed the MOMENT alembic imports this file -- no __main__ guard, by alembic's own design
)
 
 
def bootstrap(url: str, project_dir: str) -> Config:  # => co-19: init + env.py rewrite, shared setup for every Alembic example
    script_dir = os.path.join(project_dir, "migrations")  # => alembic's convention -- a "migrations" subfolder
    cfg = Config(os.path.join(project_dir, "alembic.ini"))  # => co-19: the Config object every command function needs
    cfg.set_main_option("script_location", script_dir)  # => tells Config where the scaffolded files live
    cfg.set_main_option("sqlalchemy.url", url)  # => co-19: the ONE connection string every migration runs against
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Creating directory ... done" -- path varies per run
        command.init(cfg, script_dir)  # => co-19: scaffolds env.py, script.py.mako, README, versions/ (Example 48)
    with open(os.path.join(script_dir, "env.py"), "w") as f:  # => co-19: OVERWRITES the scaffolded env.py with our own
        f.write(ENV_PY)  # => a hand-edit step, exactly like a real project's post-init setup
    return cfg  # => ready for command.revision()/upgrade()/downgrade() calls
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine, used both to reset the schema and to inspect it later
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema -- alembic's migration builds INTO this
 
    project_dir = tempfile.mkdtemp(prefix="alembic_ex49_")  # => a fresh, throwaway migration project for this example
    cfg = bootstrap(SQLA_URL, project_dir)  # => scaffolds + rewires env.py, matching Example 48's own setup
 
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Generating .../0001_create_widget.py ... done"
        rev = command.revision(cfg, message="create widget table", rev_id="0001")  # => co-19: an EMPTY skeleton, upgrade/downgrade both `pass`
    assert rev is not None and not isinstance(rev, list)  # => narrows the Union return type for pyright --strict below
    with open(rev.path, "w") as f:  # => co-19: HAND-WRITES the real migration body -- this is what "first migration" means
        f.write(  # => co-19: a real create-table upgrade, paired with its own drop-table downgrade
            '"""create widget table"""\n'  # => the migration's own docstring -- what `alembic history` displays per revision
            "from typing import Sequence, Union\n"  # => typed to match what command.revision()'s own skeleton generates
            "from alembic import op\n"  # => co-19: `op` is alembic's schema-editing API -- create_table, drop_table, add_column, etc.
            "import sqlalchemy as sa\n\n"  # => column TYPES (sa.Integer, sa.String) for the create_table() call below
            'revision: str = "0001"\n'  # => co-19: THIS revision's own id -- matches the rev_id passed to command.revision()
            "down_revision: Union[str, Sequence[str], None] = None\n"  # => co-19: None -- this is the FIRST revision, nothing before it
            "branch_labels: Union[str, Sequence[str], None] = None\n"  # => unused here -- relevant only for branching migration histories
            "depends_on: Union[str, Sequence[str], None] = None\n\n"  # => unused here -- cross-branch dependency declarations
            "def upgrade() -> None:\n"  # => co-19: the FORWARD direction -- what `alembic upgrade head` runs
            '    op.create_table("widget", sa.Column("id", sa.Integer, primary_key=True), sa.Column("name", sa.String, nullable=False))\n\n'  # => the actual DDL
            "def downgrade() -> None:\n"  # => co-19: the REVERSE direction -- what `alembic downgrade -1` runs
            '    op.drop_table("widget")\n'  # => co-19: a TRUE inverse of upgrade() -- Example 50 exercises this exact downgrade
        )
 
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows alembic's own logging-free but still noisy status lines
        command.upgrade(cfg, "head")  # => co-19: runs upgrade() from EVERY unapplied revision, in order -- just one here
 
    tables = inspect(engine).get_table_names()  # => co-19: reads Postgres' OWN catalog, not the migration file's claims
    print(f"tables after upgrade: {sorted(tables)}")  # => Output: tables after upgrade: ['alembic_version', 'widget']
    assert sorted(tables) == ["alembic_version", "widget"]  # => co-19: `widget` is REAL, plus alembic's own bookkeeping table
    # => co-19: `alembic_version` is alembic's own tracking table -- it stores exactly ONE row, the current revision id,
    # => so the NEXT `alembic upgrade`/`alembic downgrade` on this database knows precisely where it left off
 
    shutil.rmtree(project_dir)  # => cleanup -- this example's own throwaway project, not a real repository
    print("ex-49 OK")  # => Output: ex-49 OK

tables after upgrade: ['alembic_version', 'widget']
ex-49 OK

# pyright: strict
"""Example 50: upgrade() Then downgrade() -- the Schema Round-Trips Back to Empty."""
 
from __future__ import annotations
 
import contextlib  # => co-19: swallows alembic's own non-deterministic scaffolding chatter, same reason as Example 48
import io  # => the throwaway buffer redirect_stdout writes that chatter INTO
import os  # => reads connection settings, and builds/removes the scratch project directory
import shutil  # => co-19: cleanup -- removes the scaffolded project directory once this example is done
import tempfile  # => co-19: a fresh, self-contained directory for the scaffolded migration project
 
from alembic import command  # => co-19: init/revision/upgrade/downgrade -- the same API the alembic CLI itself calls
from alembic.config import Config  # => co-19: every alembic.command function needs one of these as its first argument
from sqlalchemy import create_engine, inspect, text  # => co-21: inspect() confirms the schema at EACH point in the round trip
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
# => co-19: the SAME minimal, deterministic env.py pattern Example 49 established -- no fileConfig(), no logging noise
ENV_PY = (  # => co-19: built as SEPARATE joined string lines, so each line can carry its own comment
    "from alembic import context\n"  # => co-19: `context` is alembic's OWN migration-runtime handle, not SQLAlchemy's
    "from sqlalchemy import engine_from_config, pool\n\n"  # => co-19: builds a real Engine straight from alembic.ini's own section
    "config = context.config\n"  # => co-19: the SAME Config object bootstrap() constructed, now read back inside env.py
    'target_metadata = config.attributes.get("target_metadata")\n\n\n'  # => None here -- this example writes migrations by hand
    "def run_migrations_online() -> None:\n"  # => co-19: THE function alembic's runtime calls for a live-connection migration
    "    connectable = engine_from_config(\n"  # => co-19: reads the sqlalchemy.* keys straight out of alembic.ini's own section
    '        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool\n'  # => no reuse needed
    "    )\n"  # => a short-lived migration run has no reason to KEEP a pooled connection open afterward
    "    with connectable.connect() as connection:\n"  # => borrows ONE connection for the whole migration run
    "        context.configure(connection=connection, target_metadata=target_metadata)\n"  # => co-19: binds THIS run's connection
    "        with context.begin_transaction():\n"  # => co-19: wraps every upgrade()/downgrade() call in ONE transaction
    "            context.run_migrations()\n\n\n"  # => co-19: the actual dispatch -- calls upgrade()/downgrade() on each revision
    "run_migrations_online()\n"  # => co-19: executed the MOMENT alembic imports this file -- no __main__ guard, by design
)
 
 
def bootstrap(url: str, project_dir: str) -> Config:  # => co-19: init + env.py rewrite, shared setup for every Alembic example
    script_dir = os.path.join(project_dir, "migrations")  # => alembic's convention -- a "migrations" subfolder
    cfg = Config(os.path.join(project_dir, "alembic.ini"))  # => co-19: the Config object every command function needs
    cfg.set_main_option("script_location", script_dir)  # => tells Config where the scaffolded files live
    cfg.set_main_option("sqlalchemy.url", url)  # => co-19: the ONE connection string every migration runs against
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Creating directory ... done" -- path varies per run
        command.init(cfg, script_dir)  # => co-19: scaffolds env.py, script.py.mako, README, versions/ (Example 48)
    with open(os.path.join(script_dir, "env.py"), "w") as f:  # => co-19: OVERWRITES the scaffolded env.py with our own
        f.write(ENV_PY)  # => a hand-edit step, exactly like a real project's post-init setup
    return cfg  # => ready for command.revision()/upgrade()/downgrade() calls
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine, used both to reset the schema and to inspect it later
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- fully isolated from other examples
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema -- alembic's migration builds INTO this
 
    project_dir = tempfile.mkdtemp(prefix="alembic_ex50_")  # => a fresh, throwaway migration project for this example
    # => the round trip below is exactly what a CI job or a local `alembic upgrade head && alembic downgrade base` proves
    cfg = bootstrap(SQLA_URL, project_dir)  # => scaffolds + rewires env.py, matching Examples 48-49's own setup
 
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Generating .../0001_create_widget.py ... done"
        rev = command.revision(cfg, message="create widget table", rev_id="0001")  # => co-19: an EMPTY skeleton
    assert rev is not None and not isinstance(rev, list)  # => narrows the Union return type for pyright --strict below
    with open(rev.path, "w") as f:  # => co-21: a REAL, REVERSIBLE migration -- upgrade() and downgrade() are TRUE inverses
        f.write(  # => the same create-table/drop-table pair from Example 49, reused here for the round-trip
            '"""create widget table"""\n'  # => the migration's own docstring -- what `alembic history` displays per revision
            "from typing import Sequence, Union\n"  # => typed to match what command.revision()'s own skeleton generates
            "from alembic import op\n"  # => co-19: `op` is alembic's schema-editing API
            "import sqlalchemy as sa\n\n"  # => column TYPES (sa.Integer, sa.String) for the create_table() call below
            'revision: str = "0001"\n'  # => co-19: THIS revision's own id -- matches the rev_id passed to command.revision()
            "down_revision: Union[str, Sequence[str], None] = None\n"  # => None -- this is the FIRST revision, nothing before it
            "branch_labels: Union[str, Sequence[str], None] = None\n"  # => unused here -- relevant only for branching histories
            "depends_on: Union[str, Sequence[str], None] = None\n\n"  # => unused here -- cross-branch dependency declarations
            "def upgrade() -> None:\n"  # => co-19: the FORWARD direction -- what `alembic upgrade head` runs
            '    op.create_table("widget", sa.Column("id", sa.Integer, primary_key=True), sa.Column("name", sa.String, nullable=False))\n\n'  # => DDL
            "def downgrade() -> None:\n"  # => co-21: the REVERSE direction -- what THIS example proves actually undoes the upgrade
            '    op.drop_table("widget")\n'  # => co-21: a TRUE inverse -- no data or structure survives that upgrade() didn't add
        )
 
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows alembic's own noisy status lines
        command.upgrade(cfg, "head")  # => co-19: STEP 1 of the round trip -- runs upgrade(), the schema now has `widget`
    tables_up = sorted(inspect(engine).get_table_names())  # => co-21: reads Postgres' OWN catalog right after upgrading
    print(f"tables after upgrade: {tables_up}")  # => Output: tables after upgrade: ['alembic_version', 'widget']
    assert tables_up == ["alembic_version", "widget"]  # => co-19: the forward direction landed correctly
 
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows alembic's own noisy status lines
        command.downgrade(cfg, "base")  # => co-21: STEP 2 of the round trip -- runs downgrade(), back to the START
    tables_down = sorted(inspect(create_engine(SQLA_URL)).get_table_names())  # => co-21: a FRESH inspector, no stale cache
    print(f"tables after downgrade: {tables_down}")  # => Output: tables after downgrade: ['alembic_version']
    assert tables_down == ["alembic_version"]  # => co-21: `widget` is GONE -- only alembic's own tracking table remains
    # => a migration that DOESN'T survive this round trip is a common source of "works on my machine" schema drift
    # => co-21: "base" means "no migrations applied" -- the schema is now back to where it started, MINUS alembic_version
    # => itself, which alembic keeps around even at base to remember it once managed this database at all
 
    shutil.rmtree(project_dir)  # => cleanup -- this example's own throwaway project, not a real repository
    print("ex-50 OK")  # => Output: ex-50 OK

tables after upgrade: ['alembic_version', 'widget']
tables after downgrade: ['alembic_version']
ex-50 OK

# pyright: strict
"""Example 51: Autogenerate -- Deriving a Migration From a Model<->Schema Diff."""
 
from __future__ import annotations
 
import contextlib  # => co-20: swallows alembic's own non-deterministic scaffolding chatter, same reason as Example 48
import io  # => the throwaway buffer redirect_stdout writes that chatter INTO
import os  # => reads connection settings, and builds/removes the scratch project directory
import shutil  # => co-20: cleanup -- removes the scaffolded project directory once this example is done
import tempfile  # => co-20: a fresh, self-contained directory for the scaffolded migration project
 
from alembic import command  # => co-20: revision(autogenerate=True) is what DIFFS the models against the live schema
from alembic.config import Config  # => co-20: every alembic.command function needs one of these as its first argument
from sqlalchemy import create_engine, text  # => co-20: the live database autogenerate compares the models AGAINST
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column  # => co-20: the SAME Mapped[] models Examples 14-15 taught
 
SQLA_URL: str = os.environ.get(  # => the SQLAlchemy dialect+driver URL, distinct from a plain DB-API DSN
    "SQLA_URL", "postgresql+psycopg://postgres:postgres@localhost:5432/orm_by_example"
)  # => override SQLA_URL in the environment to point at a different Postgres instance
 
 
class Base(DeclarativeBase):  # => co-06: every mapped class in this program shares ONE DeclarativeBase
    pass  # => carries no columns -- purely a registry root
 
 
class Widget(Base):  # => co-20: exists ONLY in Python -- the database starts with NO widget table at all
    __tablename__ = "widget"  # => the physical table name autogenerate will propose creating
    id: Mapped[int] = mapped_column(primary_key=True)  # => auto-assigned by Postgres
    name: Mapped[str]  # => a required TEXT column
 
 
# => co-20: the SAME minimal env.py pattern from Examples 49-50, but `target_metadata` now reads a REAL MetaData
# => object -- autogenerate needs this to know what your models SAY the schema should look like
ENV_PY = (  # => co-20: built as SEPARATE joined string lines, so each line can carry its own comment
    "from alembic import context\n"  # => co-20: `context` is alembic's OWN migration-runtime handle
    "from sqlalchemy import engine_from_config, pool\n\n"  # => co-20: builds a real Engine straight from alembic.ini's own section
    "config = context.config\n"  # => co-20: the SAME Config object bootstrap() constructed, now read back inside env.py
    'target_metadata = config.attributes.get("target_metadata")\n\n\n'  # => co-20: THIS run sets it to Widget's own Base.metadata
    "def run_migrations_online() -> None:\n"  # => co-20: THE function alembic's runtime calls for a live-connection migration
    "    connectable = engine_from_config(\n"  # => co-20: reads the sqlalchemy.* keys straight out of alembic.ini's own section
    '        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool\n'  # => no reuse needed
    "    )\n"  # => a short-lived migration run has no reason to KEEP a pooled connection open afterward
    "    with connectable.connect() as connection:\n"  # => borrows ONE connection -- autogenerate reflects THROUGH it
    "        context.configure(connection=connection, target_metadata=target_metadata)\n"  # => co-20: what autogenerate DIFFS against
    "        with context.begin_transaction():\n"  # => co-20: wraps every upgrade()/downgrade() call in ONE transaction
    "            context.run_migrations()\n\n\n"  # => the actual dispatch -- also used by autogenerate's own reflection pass
    "run_migrations_online()\n"  # => co-20: executed the MOMENT alembic imports this file -- no __main__ guard, by design
)
 
 
def bootstrap(url: str, project_dir: str, metadata: object) -> Config:  # => co-20: metadata drives WHAT autogenerate compares to
    script_dir = os.path.join(project_dir, "migrations")  # => alembic's convention -- a "migrations" subfolder
    cfg = Config(os.path.join(project_dir, "alembic.ini"))  # => co-20: the Config object every command function needs
    cfg.set_main_option("script_location", script_dir)  # => tells Config where the scaffolded files live
    cfg.set_main_option("sqlalchemy.url", url)  # => co-20: the ONE connection string every migration runs against
    with contextlib.redirect_stdout(io.StringIO()):  # => swallows "Creating directory ... done" -- path varies per run
        command.init(cfg, script_dir)  # => co-20: scaffolds env.py, script.py.mako, README, versions/ (Example 48)
    with open(os.path.join(script_dir, "env.py"), "w") as f:  # => co-20: OVERWRITES the scaffolded env.py with our own
        f.write(ENV_PY)  # => a hand-edit step, exactly like a real project's post-init setup
    cfg.attributes["target_metadata"] = metadata  # => co-20: passed to env.py via `config.attributes`, alembic's own escape hatch
    return cfg  # => ready for command.revision(autogenerate=True)
 
 
if __name__ == "__main__":  # => module entry point -- only runs when executed directly, not on import
    engine = create_engine(SQLA_URL)  # => an ORM-capable engine, used both to reset the schema and inspect the migration
    with engine.begin() as conn:  # => begin(): auto-commits on a clean exit, auto-rolls-back on an exception
        conn.execute(text("DROP SCHEMA public CASCADE"))  # => wipes EVERY table -- the database starts EMPTY, no widget yet
        conn.execute(text("CREATE SCHEMA public"))  # => a blank public schema -- autogenerate compares Widget AGAINST this
 
    project_dir = tempfile.mkdtemp(prefix="alembic_ex51_")  # => a fresh, throwaway migration project for this example
    cfg = bootstrap(SQLA_URL, project_dir, Base.metadata)  # => co-20: Widget's OWN metadata is what gets diffed below
 
    with contextlib.redirect_stdout(io.StringIO()) as buf:  # => captures the "Generating ..." AND any diff summary text
        rev = command.revision(cfg, message="add widget", autogenerate=True, rev_id="0001")  # => co-20: THE diff step itself
    assert rev is not None and not isinstance(rev, list)  # => narrows the Union return type for pyright --strict below
    generated_body = open(rev.path).read()  # => co-20: reads back the FILE autogenerate actually wrote, not a guess
    print(f"contains create_table widget: {'create_table' in generated_body and 'widget' in generated_body}")
    assert "op.create_table(" in generated_body and "'widget'" in generated_body  # => Output: contains create_table widget: True
    assert "commands auto generated by Alembic" in generated_body  # => co-20: alembic's own marker, warning "please adjust!"
    _ = buf.getvalue()  # => discarded -- captured only to keep the demonstrated stdout deterministic, not for its own content
    # => co-20: autogenerate DIFFED the live (empty) database against Widget's Mapped[] columns and proposed the exact
    # => DDL needed to reconcile them -- but it is a PROPOSAL, not a guarantee: Example 52 shows a case autogenerate
    # => gets subtly wrong, which is why every autogenerated migration still needs a human review before it ships
 
    shutil.rmtree(project_dir)  # => cleanup -- this example's own throwaway project, not a real repository
    print("ex-51 OK")  # => Output: ex-51 OK