Dlaczego zapytanie wywołuje automatyczne płukanie w SQLAlchemy?

Question

Kod widać powyżej jest tylko próbka ale działa odtworzyć ten błąd:

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

mapowanym instancja jest dodawany do sesji. Instancja chce wiedzieć (co oznacza zapytanie w bazie danych), jeśli istnieją inne instancje własnego typu o tych samych wartościach. Istnieje drugi atrybut/kolumna (_nn). Jest określony na NOT NULL. Ale domyślnie jest to NULL.

Gdy instancja (podobnie jak w próbce) jest nadal dodawana do sesji, wywołanie automatycznej przepłukiwania wywołuje query.one(). Ten flush tworzy INSERT, który próbuje zapisać instancję. To się nie powiedzie, ponieważ _nn nadal ma wartość NULL i narusza ograniczenie NOT NULL.

To właśnie rozumiem obecnie. Ale pytanie brzmi: dlaczego wywołuje auto-flush? Czy mogę to zablokować?

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

oczywiście rozwiązaniem byłoby nie dodać wystąpienie na sesji przed query.one() nazwano. Ta praca. Ale w moim rzeczywistym (ale skomplikowanym na to pytanie) przypadku użycia nie jest to miłe rozwiązanie.

Answer 1

Jak wyłączyć autoflush cechę:

• tymczasowa: można używać no_autoflush menedżera kontekstowe na fragmencie, w którym zapytanie do bazy danych, czyli w X.test metody:

  def test(self, session): 
      with session.no_autoflush: 
       q = session.query(X).filter(X._val == self._val) 
       x = q.one() 
       print('x={}'.format(x)) 
  

• sesji szeroki : wystarczy przekazać autoflush=False swojemu operatorowi sesji:

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