view raw
buhtz buhtz - 11 months ago 89
SQL Question

Why does a query invoke a auto-flush in SQLAlchemy?

The code you see above is just a sample but it works to reproduce this error:

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

A mapped instance is still added to a session. The instance wants to know (which means query on the database) if other instances its own type exists having the same values. There is a second attribute/column (
). It is specified to
. But by default it is

When the instance (like in the sample) is still added to the session a call to
invoke a auto-flush. This flush create an
which tries to store the instance. This fails because
is still null and violates the

That is what I understand currently.
But the question is why does it invoke an auto-flush? Can I block that?

#!/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 =

dbfile = 'x.db'

def _create_database():
if os.path.exists(dbfile):

engine = sa.create_engine('sqlite:///{}'.format(dbfile), echo=True)
return sao.sessionmaker(bind=engine)()

if __name__ == '__main__':
session = _create_database()

for val in range(3):
x = X(val)
x._nn = 0

x = X(1)

Of course a solution would be to not add the instance to the session before
was called. This work. But in my real (but to complex for this question) use-case it isn't a nice solution.


How to turn off autoflush feature:

  • Temporary: you can use no_autoflush context manager on snippet where you query the database, i.e. in X.test method:

    def test(self, session):
        with session.no_autoflush:
            q = session.query(X).filter(X._val == self._val)
            x =
  • Session-wide: just pass autoflush=False to your sessionmaker:

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