Complete worked example
python
from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Role(Base):
__tablename__ = "roles"
id: Mapped[int] = mapped_column(primary_key=True)
score: Mapped[int]
from sqlalchemy import update
def inspect_example(scores, threshold, increment):
engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
try:
Base.metadata.create_all(engine)
with Session(engine) as session, session.begin():
session.add_all(Role(id=i, score=value) for i, value in enumerate(scores, 1))
with Session(engine) as session, session.begin():
loaded = session.scalars(select(Role).order_by(Role.id)).all()
statement = update(Role).where(Role.score < threshold).values(score=Role.score + increment)
session.execute(statement, execution_options={"synchronize_session": "fetch"})
objects = [role.score for role in loaded]
database = list(session.scalars(select(Role.score).order_by(Role.id)))
return {"objects": objects, "database": database}
finally:
engine.dispose()- Reasoning
- The worked UPDATE uses a column predicate and the expression Role.score + increment. Session.execute receives synchronize_session set to fetch so affected identities can be synchronized. For scores [1, 5, 9], threshold 5 and increment 10, both report paths must yield [11, 5, 9]. The database observation selects score columns rather than simply returning the same loaded objects again. This matters because a second entity query can resolve through existing identities. The statement and its transaction still have clear owners; synchronization does not independently commit the database change or replace the need to handle an exception.