Performance, transactions, and migrations

from sqlalchemy import bindparam, lambda_stmt
from sqlalchemy.orm import load_only, selectinload

stmt = (
    select(User)
    .options(load_only(User.id, User.name), selectinload(User.addresses))
    .execution_options(yield_per=500)
)
for user in session.scalars(stmt):
    export(user)

by_email = lambda_stmt(
    lambda: select(User).where(User.email == bindparam("email"))
)
user = session.scalar(by_email, {"email": "ada@example.com"})

SELECT id, name FROM user_account;
SELECT * FROM address WHERE user_id IN (...);
SELECT * FROM user_account WHERE email = :email;

import logging
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
# Or create_engine(url, echo="debug") in development.

plan = session.execute(
    text("EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM user_account WHERE email=:e"),
    {"e": "ada@example.com"},
).scalars().all()

EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM user_account WHERE email = 'ada@example.com';

from sqlalchemy import update

with SessionLocal.begin() as session:
    result = session.execute(
        update(Product)
        .where(Product.id == product_id, Product.quantity >= requested)
        .values(quantity=Product.quantity - requested)
    )
    if result.rowcount != 1:
        raise ValueError("out of stock or missing product")

UPDATE product
SET quantity = quantity - :requested
WHERE id = :product_id AND quantity >= :requested;
-- One row means the rule held at write time.

from sqlalchemy.exc import IntegrityError, OperationalError

try:
    with SessionLocal.begin() as session:
        session.add(User(name="Ada", email="taken@example.com"))
except IntegrityError as exc:
    # Context manager rolled back already.
    raise ValueError("email is already registered") from exc

# OperationalError can include timeout/disconnect/deadlock.
# Retry only documented transient failures and retry the WHOLE
# idempotent transaction with a fresh Session.

INSERT INTO user_account (name, email)
VALUES ('Ada', 'taken@example.com');
-- UNIQUE violation
ROLLBACK;

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

@pytest.fixture
def session():
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
        session.rollback()
    engine.dispose()

def test_email_unique(session):
    session.add(User(name="Ada", email="ada@example.com"))
    session.commit()
    session.add(User(name="Other", email="ada@example.com"))
    with pytest.raises(IntegrityError):
        session.commit()

pip install alembic
alembic init migrations
# Set target_metadata = Base.metadata in env.py
alembic revision --autogenerate -m "create user account"
alembic upgrade head
alembic downgrade -1
alembic current
alembic history

def upgrade() -> None:
    op.add_column(
        "user_account",
        sa.Column("nickname", sa.String(), nullable=True),
    )
    op.create_index("ix_user_account_name", "user_account", ["name"])

def downgrade() -> None:
    op.drop_index("ix_user_account_name", table_name="user_account")
    op.drop_column("user_account", "nickname")

ALTER TABLE user_account ADD COLUMN nickname VARCHAR;
CREATE INDEX ix_user_account_name ON user_account (name);