Atomic stock decrement and locking in SQLAlchemy (update, with_for_update, version_id_col)

# ======================================================================
# Q: How do you do an atomic stock decrement and locking in SQLAlchemy?
#
# The same three solutions I know from Django (chapter 4). The difference is
# that optimistic locking is built into SQLAlchemy with version_id_col.
# ======================================================================

from sqlalchemy import update

# 1) Atomic update in the database (like F() in Django)
res = session.execute(
    update(Product)
    .where(Product.id == pid, Product.stock >= qty)
    .values(stock=Product.stock - qty))
if res.rowcount == 0:
    raise OutOfStock(pid)

# 2) Pessimistic lock (like select_for_update)
product = session.scalars(
    select(Product).where(Product.id == pid).with_for_update()).one()

# 3) Optimistic locking: built in
class Product(Base):
    version: Mapped[int] = mapped_column(default=1)
    __mapper_args__ = {'version_id_col': version}   # StaleDataError on conflict