您在上面看到的代码只是一个示例,但是可以重现此错误:
sqlalchemy.exc.IntegrityError: (raised as a result of Query-invoked autoflush; consider using a session.no_autoflush block if this flush is occurring prematurely) (sqlite3.IntegrityError) NOT NULL constraint failed: X.nn [SQL: 'INSERT INTO "X" (nn, val) VALUES (?, ?)'] [parameters: (None, 1)]
映射的实例仍添加到会话中。该实例想知道(这意味着在数据库上查询)是否存在其他自己实例类型具有相同值的实例。还有第二个属性/列(_nn)。它指定为NOT NULL。但默认情况下为NULL。
_nn
NOT NULL
NULL
当实例(如示例中)仍添加到会话中时,将query.one()调用一个调用以自动刷新。刷新创建一个INSERT尝试存储实例的。之所以失败,_nn是因为它仍然为null并违反了NOT NULL约束。
query.one()
INSERT
这就是我目前所了解的。但是问题是为什么它要调用自动冲洗?我可以阻止吗?
#!/usr/bin/env python3 import os.path import os import sqlalchemy as sa import sqlalchemy.orm as sao import sqlalchemy.ext.declarative as sad from sqlalchemy_utils import create_database _Base = sad.declarative_base() session = None class X(_Base): __tablename__ = 'X' _oid = sa.Column('oid', sa.Integer, primary_key=True) _nn = sa.Column('nn', sa.Integer, nullable=False) # NOT NULL! _val = sa.Column('val', sa.Integer) def __init__(self, val): self._val = val def test(self, session): q = session.query(X).filter(X._val == self._val) x = q.one() print('x={}'.format(x)) dbfile = 'x.db' def _create_database(): if os.path.exists(dbfile): os.remove(dbfile) engine = sa.create_engine('sqlite:///{}'.format(dbfile), echo=True) create_database(engine.url) _Base.metadata.create_all(engine) return sao.sessionmaker(bind=engine)() if __name__ == '__main__': session = _create_database() for val in range(3): x = X(val) x._nn = 0 session.add(x) session.commit() x = X(1) session.add(x) x.test(session)
当然,解决方案是在调用之前 不 将实例添加到会话query.one()中。这项工作。但是在我的实际(但是对于这个问题来说很复杂)的用例中,这不是一个很好的解决方案。
如何关闭自动冲洗功能:
临时的:您可以在查询数据库的代码段中使用no_autoflush上下文管理器,即在X.test方法中:
X.test
def test(self, session): with session.no_autoflush: q = session.query(X).filter(X._val == self._val) x = q.one() print('x={}'.format(x))
autoflush=False全会期:只需传递给您的会议主持人即可:
autoflush=False
return sao.sessionmaker(bind=engine, autoflush=False)()