Behavioral Changes
==================
+.. _legacy_is_orphan_addition:
+
+The consideration of a "pending" object as an "orphan" has been made more aggressive
+------------------------------------------------------------------------------------
+
+This is a late add to the 0.8 series, however it is hoped that the new behavior
+is generally more consistent and intuitive in a wider variety of
+situations. The ORM has since at least version 0.4 included behavior
+such that an object that's "pending", meaning that it's
+associated with a :class:`.Session` but hasn't been inserted into the database
+yet, is automatically expunged from the :class:`.Session` when it becomes an "orphan",
+which means it has been de-associated with a parent object that refers to it
+with ``delete-orphan`` cascade on the configured :func:`.relationship`. This
+behavior is intended to approximately mirror the behavior of a persistent
+(that is, already inserted) object, where the ORM will emit a DELETE for such
+objects that become orphans based on the interception of detachment events.
+
+The behavioral change comes into play for objects that
+are referred to by multiple kinds of parents that each specify ``delete-orphan``; the
+typical example is an :ref:`association object <association_pattern>` that bridges two other kinds of objects
+in a many-to-many pattern. Previously, the behavior was such that the
+pending object would be expunged only when de-associated with *all* of its parents.
+With the behavioral change, the pending object
+is expunged as soon as it is de-associated from *any* of the parents that it was
+previously associated with. This behavior is intended to more closely
+match that of persistent objects, which are deleted as soon
+as they are de-associated from any parent.
+
+The rationale for the older behavior dates back
+at least to version 0.4, and was basically a defensive decision to try to alleviate
+confusion when an object was still being constructed for INSERT. But the reality
+is that the object is re-associated with the :class:`.Session` as soon as it is
+attached to any new parent in any case.
+
+It's still possible to flush an object
+that is not associated with all of its required parents, if the object was either
+not associated with those parents in the first place, or if it was expunged, but then
+re-associated with a :class:`.Session` via a subsequent attachment event but still
+not fully associated. In this situation, it is expected that the database
+would emit an integrity error, as there are likely NOT NULL foreign key columns
+that are unpopulated. The ORM makes the decision to let these INSERT attempts
+occur, based on the judgment that an object that is only partially associated with
+its required parents but has been actively associated with some of them,
+is more often than not a user error, rather than an intentional
+omission which should be silently skipped - silently skipping the INSERT here would
+make user errors of this nature very hard to debug.
+
+The old behavior, for applications that might have been relying upon it, can be re-enabled for
+any :class:`.Mapper` by specifying the flag ``legacy_is_orphan`` as a mapper
+option.
+
+The new behavior allows the following test case to work::
+
+ from sqlalchemy import Column, Integer, String, ForeignKey
+ from sqlalchemy.orm import relationship, backref
+ from sqlalchemy.ext.declarative import declarative_base
+
+ Base = declarative_base()
+
+ class User(Base):
+ __tablename__ = 'user'
+ id = Column(Integer, primary_key=True)
+ name = Column(String(64))
+
+ class UserKeyword(Base):
+ __tablename__ = 'user_keyword'
+ user_id = Column(Integer, ForeignKey('user.id'), primary_key=True)
+ keyword_id = Column(Integer, ForeignKey('keyword.id'), primary_key=True)
+
+ user = relationship(User,
+ backref=backref("user_keywords",
+ cascade="all, delete-orphan")
+ )
+
+ keyword = relationship("Keyword",
+ backref=backref("user_keywords",
+ cascade="all, delete-orphan")
+ )
+
+ # uncomment this to enable the old behavior
+ # __mapper_args__ = {"legacy_is_orphan": True}
+
+ class Keyword(Base):
+ __tablename__ = 'keyword'
+ id = Column(Integer, primary_key=True)
+ keyword = Column('keyword', String(64))
+
+ from sqlalchemy import create_engine
+ from sqlalchemy.orm import Session
+
+ # note we're using Postgresql to ensure that referential integrity
+ # is enforced, for demonstration purposes.
+ e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
+
+ Base.metadata.drop_all(e)
+ Base.metadata.create_all(e)
+
+ session = Session(e)
+
+ u1 = User(name="u1")
+ k1 = Keyword(keyword="k1")
+
+ session.add_all([u1, k1])
+
+ uk1 = UserKeyword(keyword=k1, user=u1)
+
+ # previously, if session.flush() were called here,
+ # this operation would succeed, but if session.flush()
+ # were not called here, the operation fails with an
+ # integrity error.
+ # session.flush()
+ del u1.user_keywords[0]
+
+ session.commit()
+
+
+:ticket:`2655`
+
The after_attach event fires after the item is associated with the Session instead of before; before_attach added
-----------------------------------------------------------------------------------------------------------------
from sqlalchemy.orm import mapper, relationship, create_session, \
sessionmaker, class_mapper, backref, Session, util as orm_util,\
configure_mappers
-from sqlalchemy.orm import attributes, exc as orm_exc
+from sqlalchemy.orm.attributes import instance_state
+from sqlalchemy.orm import attributes, exc as orm_exc, object_mapper
from sqlalchemy import testing
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
Column('account_id', Integer,
ForeignKey('accounts.account_id')))
- def test_double_parent_expunge_o2m(self):
- """test the delete-orphan uow event for multiple delete-orphan
- parent relationships."""
-
+ def _fixture(self, legacy_is_orphan, uselist):
sales_reps, customers, accounts = (self.tables.sales_reps,
self.tables.customers,
self.tables.accounts)
class SalesRep(fixtures.ComparableEntity):
pass
- mapper(Customer, customers)
+ mapper(Customer, customers, legacy_is_orphan=legacy_is_orphan)
mapper(Account, accounts, properties=dict(
customers=relationship(Customer,
cascade="all,delete-orphan",
- backref="account")))
+ backref="account",
+ uselist=uselist)))
mapper(SalesRep, sales_reps, properties=dict(
customers=relationship(Customer,
cascade="all,delete-orphan",
- backref="sales_rep")))
+ backref="sales_rep",
+ uselist=uselist)))
s = create_session()
a = Account(balance=0)
c = Customer(name="Jane")
- a.customers.append(c)
- sr.customers.append(c)
+ if uselist:
+ a.customers.append(c)
+ sr.customers.append(c)
+ else:
+ a.customers = c
+ sr.customers = c
+
assert c in s
+ return s, c, a, sr
+
+ def test_double_parent_expunge_o2m_legacy(self):
+ """test the delete-orphan uow event for multiple delete-orphan
+ parent relationships."""
+
+ s, c, a, sr = self._fixture(True, True)
a.customers.remove(c)
assert c in s, "Should not expunge customer yet, still has one parent"
assert c not in s, \
'Should expunge customer when both parents are gone'
- def test_double_parent_expunge_o2o(self):
+ def test_double_parent_expunge_o2m_current(self):
"""test the delete-orphan uow event for multiple delete-orphan
parent relationships."""
- sales_reps, customers, accounts = (self.tables.sales_reps,
- self.tables.customers,
- self.tables.accounts)
-
+ s, c, a, sr = self._fixture(False, True)
- class Customer(fixtures.ComparableEntity):
- pass
- class Account(fixtures.ComparableEntity):
- pass
- class SalesRep(fixtures.ComparableEntity):
- pass
+ a.customers.remove(c)
+ assert c not in s, "Should expunge customer when either parent is gone"
- mapper(Customer, customers)
- mapper(Account, accounts, properties=dict(
- customer=relationship(Customer,
- cascade="all,delete-orphan",
- backref="account", uselist=False)))
- mapper(SalesRep, sales_reps, properties=dict(
- customer=relationship(Customer,
- cascade="all,delete-orphan",
- backref="sales_rep", uselist=False)))
- s = create_session()
+ sr.customers.remove(c)
+ assert c not in s, \
+ 'Should expunge customer when both parents are gone'
- a = Account(balance=0)
- sr = SalesRep(name="John")
- s.add_all((a, sr))
- s.flush()
+ def test_double_parent_expunge_o2o_legacy(self):
+ """test the delete-orphan uow event for multiple delete-orphan
+ parent relationships."""
- c = Customer(name="Jane")
+ s, c, a, sr = self._fixture(True, False)
- a.customer = c
- sr.customer = c
- assert c in s
- a.customer = None
+ a.customers = None
assert c in s, "Should not expunge customer yet, still has one parent"
- sr.customer = None
+ sr.customers = None
+ assert c not in s, \
+ 'Should expunge customer when both parents are gone'
+
+ def test_double_parent_expunge_o2o_current(self):
+ """test the delete-orphan uow event for multiple delete-orphan
+ parent relationships."""
+
+ s, c, a, sr = self._fixture(False, False)
+
+
+ a.customers = None
+ assert c not in s, "Should expunge customer when either parent is gone"
+
+ sr.customers = None
assert c not in s, \
'Should expunge customer when both parents are gone'
eq_(sess.query(A).get(a1.id),
A(name='a1', bs=[B(name='b1'), B(name='b2'), B(name='b3')]))
+class OrphanCriterionTest(fixtures.MappedTest):
+ @classmethod
+ def define_tables(self, metadata):
+ Table("core", metadata,
+ Column("id", Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ Column("related_one_id", Integer, ForeignKey("related_one.id")),
+ Column("related_two_id", Integer, ForeignKey("related_two.id"))
+ )
+
+ Table("related_one", metadata,
+ Column("id", Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ )
+
+ Table("related_two", metadata,
+ Column("id", Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ )
+
+ def _fixture(self, legacy_is_orphan, persistent,
+ r1_present, r2_present, detach_event=True):
+ class Core(object):
+ pass
+
+ class RelatedOne(object):
+ def __init__(self, cores):
+ self.cores = cores
+
+ class RelatedTwo(object):
+ def __init__(self, cores):
+ self.cores = cores
+
+ mapper(Core, self.tables.core, legacy_is_orphan=legacy_is_orphan)
+ mapper(RelatedOne, self.tables.related_one, properties={
+ 'cores': relationship(Core, cascade="all, delete-orphan",
+ backref="r1")
+ })
+ mapper(RelatedTwo, self.tables.related_two, properties={
+ 'cores': relationship(Core, cascade="all, delete-orphan",
+ backref="r2")
+ })
+ c1 = Core()
+ if detach_event:
+ r1 = RelatedOne(cores=[c1])
+ r2 = RelatedTwo(cores=[c1])
+ else:
+ if r1_present:
+ r1 = RelatedOne(cores=[c1])
+ if r2_present:
+ r2 = RelatedTwo(cores=[c1])
+
+ if persistent:
+ s = Session()
+ s.add(c1)
+ s.flush()
+
+ if detach_event:
+ if not r1_present:
+ c1.r1 = None
+ if not r2_present:
+ c1.r2 = None
+ return c1
+
+ def _assert_not_orphan(self, c1):
+ mapper = object_mapper(c1)
+ state = instance_state(c1)
+ assert not mapper._is_orphan(state)
+
+ def _assert_is_orphan(self, c1):
+ mapper = object_mapper(c1)
+ state = instance_state(c1)
+ assert mapper._is_orphan(state)
+
+ def test_leg_pers_r1_r2(self):
+ c1 = self._fixture(True, True, True, True)
+
+ self._assert_not_orphan(c1)
+
+ def test_current_pers_r1_r2(self):
+ c1 = self._fixture(False, True, True, True)
+
+ self._assert_not_orphan(c1)
+
+ def test_leg_pers_r1_notr2(self):
+ c1 = self._fixture(True, True, True, False)
+
+ self._assert_not_orphan(c1)
+
+ def test_current_pers_r1_notr2(self):
+ c1 = self._fixture(False, True, True, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_leg_pers_notr1_notr2(self):
+ c1 = self._fixture(True, True, False, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_current_pers_notr1_notr2(self):
+ c1 = self._fixture(False, True, True, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_leg_transient_r1_r2(self):
+ c1 = self._fixture(True, False, True, True)
+
+ self._assert_not_orphan(c1)
+
+ def test_current_transient_r1_r2(self):
+ c1 = self._fixture(False, False, True, True)
+
+ self._assert_not_orphan(c1)
+
+ def test_leg_transient_r1_notr2(self):
+ c1 = self._fixture(True, False, True, False)
+
+ self._assert_not_orphan(c1)
+
+ def test_current_transient_r1_notr2(self):
+ c1 = self._fixture(False, False, True, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_leg_transient_notr1_notr2(self):
+ c1 = self._fixture(True, False, False, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_current_transient_notr1_notr2(self):
+ c1 = self._fixture(False, False, False, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_leg_transient_notr1_notr2_noevent(self):
+ c1 = self._fixture(True, False, False, False, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_current_transient_notr1_notr2_noevent(self):
+ c1 = self._fixture(False, False, False, False, False)
+
+ self._assert_is_orphan(c1)
+
+ def test_leg_persistent_notr1_notr2_noevent(self):
+ c1 = self._fixture(True, True, False, False, False)
+
+ self._assert_not_orphan(c1)
+
+ def test_current_persistent_notr1_notr2_noevent(self):
+ c1 = self._fixture(False, True, False, False, False)
+
+ self._assert_not_orphan(c1)
+
class O2MConflictTest(fixtures.MappedTest):
"""test that O2M dependency detects a change in parent, does the
right thing, and updates the collection/attribute.
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
