I have an application which listens for before_flush events and does some
side effects when column values are deleted. But in some cases the history
for an attribute is empty even though it has been deleted. For example:
person = Person(address="VALUE1")
self.session.add(person)
self.session.flush()
#self.session.commit()
# Without the next line, deletion of VALUE1 is not registered if committed
above
#assert person.address
person.address = None
self.session.commit()
Here, "VALUE1" is registered as deleted in the history only if:
- flush is called but not commit, or
- person.address is accessed before it is set to None
Is this a bug or is it expected?
A complete example is attached with some prints that show the attribute state.
SQLAlchemy version 1.1.14
python version 3.6.2 [GCC 4.2.1 Compatible Apple LLVM 8.1.0 (clang-802.0.42)]
Best regards
Levon Saldamli
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
import logging
import sys
import sqlalchemy
from sqlalchemy import Column, Integer, inspect, event
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql.sqltypes import String
logging.disable(logging.CRITICAL)
engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)
def myprint(*args, **kwargs):
print("-----", *args, **kwargs)
class SessionEventListener:
@classmethod
def _before_flush(cls, session: Session, flush_context, instances):
myprint("_before_flush")
cls._deleted_values(session)
@classmethod
def _after_flush(cls, session: Session, flush_context):
myprint("_after_flush")
cls._deleted_values(session)
@classmethod
def _after_flush_postexec(cls, session: Session, flush_context):
myprint("_after_flush_postexec")
cls._deleted_values(session)
@classmethod
def _before_commit(cls, session: Session):
myprint("_before_commit")
cls._deleted_values(session)
@classmethod
def _after_commit(cls, session: Session):
myprint("_after_commit")
cls._deleted_values(session)
@classmethod
def _deleted_values(cls, session):
added_or_dirty_models = list(session.new) + list(session.dirty)
for model in added_or_dirty_models:
for deleted_values in cls._deleted_values_in(model):
myprint(f"DELETED VALUE: {deleted_values}")
@classmethod
def _deleted_values_in(cls, model_instance):
states = cls._value_attribute_states(model_instance)
for attribute_state in states:
myprint(f"history: {attribute_state.key}: {attribute_state.history}")
deleted = attribute_state.history.deleted or []
for value in deleted:
if value is not None:
yield value
@classmethod
def _value_attribute_states(cls, model_instance):
instance_state = inspect(model_instance)
for attribute_state in instance_state.attrs:
yield attribute_state
@classmethod
def _column_is_of_type(cls, inspected_instance, column_key, wanted_type):
column = inspected_instance.mapper.columns.get(column_key)
return column is not None and isinstance(column.type, wanted_type)
@classmethod
def setup(cls, session):
event.listen(session, 'before_flush', cls._before_flush)
event.listen(session, 'after_flush', cls._after_flush)
event.listen(session, 'after_flush_postexec', cls._after_flush_postexec)
event.listen(session, 'before_commit', cls._before_commit)
event.listen(session, 'after_commit', cls._after_commit)
class Person(Base):
__tablename__ = 'container'
id = Column(Integer, primary_key=True, autoincrement=True)
address = Column(String, nullable=True, unique=True)
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
Base.metadata.create_all(engine)
class TestClass:
def __init__(self, session):
self.session = session
SessionEventListener.setup(self.session)
def test_1(self):
""" DELETED VALUE1 and DELETED VALUE2 is expected"""
person = Person(address="VALUE1")
self.session.add(person)
self.session.commit()
value_1 = person.address
myprint(f"VALUE_1: {value_1}")
person.address = "VALUE2"
self.session.flush()
self.session.commit()
# Without the next line, deletion of VALUE2 is not registered if committed above
# assert person.address
person.address = "VALUE3"
self.session.commit()
def test_2(self):
""" DELETED VALUE1 is expected"""
person = Person(address="VALUE1")
self.session.add(person)
#self.session.flush()
self.session.commit()
# Without the next line, deletion of VALUE1 is not registered if committed above
# assert person.address
person.address = None
self.session.commit()
myprint("SQLALchemy version", sqlalchemy.__version__ )
myprint("python version", sys.version )
session = Session()
test1 = TestClass(session)
myprint("")
myprint("TEST_1")
myprint("")
test1.test_1()
myprint("")
myprint("TEST_2")
myprint("")
session.close()
session = Session()
test2 = TestClass(session)
test2.test_2()
session.close()
