我正在尝试在本地重现一个错误,我认为这是由更新依赖于陈旧数据的竞争条件引起的(由于 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.
这就是我想要重现的情况。
最佳答案
上次更新查询不使用陈旧的 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()
关于python - 如何重现由 sqlalchemy session 缓存引起的错误,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57828965/