Bulk UPDATE expires server_onupdate/postfetch columns on only one matched object #13675

        to_expire = {...postfetch_cols...}.difference(evaluated_keys)   # added in 2.0.36
        for obj, state, dict_ in matched_objects:
            ...
            state._expire_attributes(state.dict, to_expire)
            ...
            to_expire = attrib.intersection(dict_).difference(to_evaluate)   # pre-existing
            if to_expire:
                state._expire_attributes(dict_, to_expire)

import sqlalchemy as sa
from sqlalchemy import FetchedValue, func, inspect, update
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column


class Base(DeclarativeBase):
    pass


class Thing(Base):
    __tablename__ = "thing"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    stamped: Mapped[str | None]
    touched: Mapped[str] = mapped_column(server_default="x", server_onupdate=FetchedValue())


engine = sa.create_engine("sqlite://")
Base.metadata.create_all(engine)
with engine.begin() as conn:
    conn.exec_driver_sql(
        "CREATE TRIGGER t AFTER UPDATE ON thing BEGIN "
        "UPDATE thing SET touched = 'y' WHERE id = NEW.id AND touched = 'x'; END"
    )

print("SQLAlchemy", sa.__version__)

# 1. A server_onupdate column is expired on only one of the matched objects.
with Session(engine) as session:
    things = [Thing(id=n, name=f"thing {n}") for n in (1, 2, 3)]
    session.add_all(things)
    session.commit()
    for thing in things:
        thing.touched  # load it: 'x'
    session.execute(update(Thing).values(name="renamed"))
    for thing in things:
        print("case 1", thing.id, "touched expired:", "touched" in inspect(thing).expired_attributes)
    print("case 1 values:", [thing.touched for thing in things], "(the database has 'y' for all)")

# 2. A SET column given a SQL expression, not yet loaded, is expired on only one object;
#    the others then read None without loading.
with Session(engine) as session:
    session.execute(sa.delete(Thing))
    things = [Thing(id=n, name=f"thing {n}") for n in (11, 12, 13)]
    session.add_all(things)
    session.flush()
    session.execute(update(Thing).values(stamped=func.upper("z")))
    for thing in things:
        print("case 2", thing.id, "stamped expired:", "stamped" in inspect(thing).expired_attributes)
    print("case 2 values:", [thing.stamped for thing in things], "(the database has 'Z' for all)")