set
)
- def _post_update(self, state, uowcommit, related):
+ def _post_update(self, state, uowcommit, related, is_m2o_delete=False):
for x in related:
- if x is not None:
+ if not is_m2o_delete or x is not None:
uowcommit.issue_post_update(
state,
[r for l, r in self.prop.synchronize_pairs]
self.key,
self._passive_delete_flag)
if history:
- self._post_update(state, uowcommit, history.sum())
+ self._post_update(
+ state, uowcommit, history.sum(),
+ is_m2o_delete=True)
def process_saves(self, uowcommit, states):
for state in states:
from sqlalchemy.testing.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, backref, \
create_session, sessionmaker
-from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_, is_
from sqlalchemy.testing.assertsql import RegexSQL, CompiledSQL, AllOf
from sqlalchemy.testing import fixtures
{'id': b4.id}])
)
+ def test_post_update_m2o_detect_none(self):
+ person, ball, Ball, Person = (
+ self.tables.person,
+ self.tables.ball,
+ self.classes.Ball,
+ self.classes.Person)
+
+ mapper(Ball, ball, properties={
+ 'person': relationship(
+ Person, post_update=True,
+ primaryjoin=person.c.id == ball.c.person_id)
+ })
+ mapper(Person, person)
+
+ sess = create_session(autocommit=False, expire_on_commit=True)
+ sess.add(Ball(person=Person()))
+ sess.commit()
+ b1 = sess.query(Ball).first()
+
+ # needs to be unloaded
+ assert 'person' not in b1.__dict__
+ b1.person = None
+
+ self.assert_sql_execution(
+ testing.db,
+ sess.flush,
+ CompiledSQL(
+ "UPDATE ball SET person_id=:person_id WHERE ball.id = :ball_id",
+ lambda ctx: {'person_id': None, 'ball_id': b1.id})
+ )
+
+ is_(b1.person, None)
+
class SelfReferentialPostUpdateTest(fixtures.MappedTest):
"""Post_update on a single self-referential mapper.