为什么查询在SQLAlchemy中调用自动刷新？

Question

您在上面看到的代码只是一个示例,但它可以重现此错误:

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)]

映射的实例仍会添加到会话中.实例想要知道(这意味着对数据库的查询)其他实例是否存在具有相同值的自己的类型.还有第二个属性/ column(_nn).它被指定为NOT NULL.但默认情况下是NULL.

当实例(如示例中)仍然添加到会话中时,调用以query.one()调用自动刷新.此flush创建一个INSERT尝试存储实例的.这失败,因为_nn仍然为null并违反NOT NULL约束.

这就是我目前所理解的.但问题是为什么它会调用自动刷新？我能阻止吗？

#!/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()中.这项工作.但在我的真实(但对于这个问题的复杂)用例,它不是一个很好的解决方案.

Answer 1

如何关闭autoflush功能:

• 临时:您可以在查询数据库的代码段上使用no_autoflush上下文管理器,即在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给您的会话制作者:

  return sao.sessionmaker(bind=engine, autoflush=False)()
  

Answer 2

我知道这已经很旧了，但对于其他在使用 Flask-sqlalchemy 时遇到此错误的人来说可能会有所帮助。下面的代码解决了我的自动刷新问题。

db = SQLAlchemy(session_options={"autoflush": False})