如何重现由 sqlalchemy 会话缓存导致的错误

Question

我正在尝试在本地重现一个错误，我认为该错误是由更新依赖于过时数据的竞争条件引起的(由于synchronize_session=False(，本质上如下所示：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, Boolean, CheckConstraint
from sqlalchemy.orm.session import sessionmaker
Base = declarative_base()
# change this to your actual postgres url
db_string = "postgres://max:steve@localhost/test"
db = create_engine(db_string)

class User(Base):
__tablename__ = 'users4'
id = Column(Integer, primary_key=True)
deleted = Column(Boolean)
super_user = Column(Boolean, CheckConstraint('NOT (super_user AND deleted)', name='check1'))

Base.metadata.create_all(db)
Session = sessionmaker(bind=db)
session = Session()
session.autoflush = False
# Create a user
session.add(User(id=1, deleted=False, super_user=False))
# Delete that user
session.query(User).filter(User.id == 1).update(
{'deleted': True}, synchronize_session=False)
# Make all non-deleted users into super users
# Will violate the CHECK constraint if it's the previous query hasn't 
# been flushed
session.query(User).filter(User.deleted == False).update({'super_user': True})

有没有办法强制 sqlalchemy 使用缓存的会话(可能通过模拟或类似的东西(，以便此代码将引发违反约束并引发IntegrityError？

synchronize_session的文档说

。更新的对象可能仍保留在会话中，其属性值过时，这可能会导致结果混乱。

这就是我想重现的情况。

Answer 1

上次更新查询不使用过时的会话数据。 我认为像这样的情况，逻辑作用于过时的属性将在最终发生刷新时触发检查约束：

# Create a user
user1 = User(id=1, deleted=False, super_user=False)
session.add(user1)

# Delete that user
session.query(User).filter(User.id == 1).update(
{'deleted': True}, synchronize_session=False)

# Make all non-deleted users into super users
# Will violate the CHECK constraint if it's the previous query hasn't 
# been flushed
if not user1.deleted:
user1.super_user = True
session.flush()