related = state.get_impl(self.key).get(state, dict_,
passive=self._passive_update_flag)
if related is not attributes.PASSIVE_NO_RESULT and \
- related is not None:
+ related is not None:
related_state = attributes.instance_state(dict_[self.key])
if related_state in switchers:
uowcommit.register_object(state,
def _synchronize(self, state, child, associationrow,
clearkeys, uowcommit, operation):
- if associationrow is None:
- return
+ # this checks for None if uselist=True
self._verify_canload(child)
+ # but if uselist=False we get here. If child is None,
+ # no association row can be generated, so return.
+ if child is None:
+ return False
+
if child is not None and not uowcommit.session._contains_state(child):
if not child.deleted:
util.warn(
)
Table('secondary', metadata,
- Column('left_id', Integer, ForeignKey('left.id')),
- Column('right_id', Integer, ForeignKey('right.id')),
+ Column('left_id', Integer, ForeignKey('left.id'),
+ primary_key=True),
+ Column('right_id', Integer, ForeignKey('right.id'),
+ primary_key=True),
)
@classmethod
})
mapper(B, right)
+ def _bidirectional_onescalar_fixture(self):
+ left, secondary, right = self.tables.left, \
+ self.tables.secondary, self.tables.right
+ A, B = self.classes.A, self.classes.B
+ mapper(A, left, properties={
+ 'bs': relationship(B, secondary=secondary,
+ backref=backref('a', uselist=False),
+ order_by=right.c.id)
+ })
+ mapper(B, right)
+
def test_session_delete(self):
self._standard_bidirectional_fixture()
A, B = self.classes.A, self.classes.B
sess.flush()
eq_(sess.query(secondary).count(), 0)
+ def test_remove_scalar(self):
+ # test setting a uselist=False to None
+ self._bidirectional_onescalar_fixture()
+ A, B = self.classes.A, self.classes.B
+ secondary = self.tables.secondary
+
+ sess = Session()
+ sess.add_all([
+ A(data='a1', bs=[B(data='b1'), B(data='b2')]),
+ ])
+ sess.commit()
+
+ a1 = sess.query(A).filter_by(data='a1').one()
+ b2 = sess.query(B).filter_by(data='b2').one()
+ assert b2.a is a1
+
+ b2.a = None
+ sess.commit()
+
+ eq_(a1.bs, [B(data='b1')])
+ eq_(b2.a, None)
+ eq_(sess.query(secondary).count(), 1)