Complete worked example
python
from pathlib import Path
from tempfile import TemporaryDirectory
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.orm.exc import StaleDataError
class Base(DeclarativeBase):
pass
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(primary_key=True)
score: Mapped[int]
version: Mapped[int] = mapped_column(default=1, nullable=False)
__mapper_args__ = {"version_id_col": version}
def inspect_example(initial, committed, late_value):
with TemporaryDirectory() as folder:
engine = create_engine("sqlite+pysqlite:///" + str(Path(folder) / "owned.sqlite3"),
connect_args={"autocommit": False})
try:
Base.metadata.create_all(engine)
with Session(engine) as seed, seed.begin():
seed.add(Role(id=1, score=initial))
with Session(engine, expire_on_commit=False) as stale:
role = stale.get(Role, 1)
stale.commit() # release the read transaction, retain this old version
with Session(engine) as writer, writer.begin():
writer.get(Role, 1).score = committed
conflict = False
try:
role.score = late_value
stale.commit()
except StaleDataError:
stale.rollback()
conflict = True
with Session(engine) as check:
current = check.get(Role, 1)
return {"score": current.score, "version": current.version, "conflict": conflict}
finally:
engine.dispose()- Reasoning
- The row starts with score 10 and version one. A second Session changes the score to 20, advancing the version. The stale object then attempts a different score using its old version and the flush is rejected. The handler rolls back before reading the final state, which remains score 20 at version two. If the intervening writer made no actual change, there may be no version advance or conflict. The cases include that distinction. Version checking concerns actual ORM writes; assigning an attribute its unchanged value need not emit an UPDATE simply because the Python assignment statement ran.