python - 如何重现由 sqlalchemy session 缓存引起的错误

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 使用缓存的 session (可能通过模拟或类似的方式)，以便此代码违反约束并引发 IntegrityError？

docs for synchronize_session这么说

... updated objects may still remain in the session with stale values on their attributes, which can lead to confusing results.

这就是我想要重现的情况。

Answer 1

上次更新查询不使用陈旧的 session 数据。我认为像这样的情况，当最终确实发生刷新时，逻辑作用于陈旧属性将触发检查约束:

# 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()