:ticket:`3582`
+.. _change_2349:
+
+passive_deletes feature for joined-inheritance mappings
+-------------------------------------------------------
+
+A joined-table inheritance mapping may now allow a DELETE to proceed
+as a result of :meth:`.Session.delete`, which only emits DELETE for the
+base table, and not the subclass table, allowing configured ON DELETE CASCADE
+to take place for the configured foreign keys. This is configured using
+the :paramref:`.orm.mapper.passive_deletes` option::
+
+ from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
+ from sqlalchemy.orm import Session
+ from sqlalchemy.ext.declarative import declarative_base
+
+ Base = declarative_base()
+
+
+ class A(Base):
+ __tablename__ = "a"
+ id = Column('id', Integer, primary_key=True)
+ type = Column(String)
+
+ __mapper_args__ = {
+ 'polymorphic_on': type,
+ 'polymorphic_identity': 'a',
+ 'passive_deletes': True
+ }
+
+
+ class B(A):
+ __tablename__ = 'b'
+ b_table_id = Column('b_table_id', Integer, primary_key=True)
+ bid = Column('bid', Integer, ForeignKey('a.id', ondelete="CASCADE"))
+ data = Column('data', String)
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'b'
+ }
+
+With the above mapping, the :paramref:`.orm.mapper.passive_deletes` option
+is configured on the base mapper; it takes effect for all non-base mappers
+that are descendants of the mapper with the option set. A DELETE for
+an object of type ``B`` no longer needs to retrieve the primary key value
+of ``b_table_id`` if unloaded, nor does it need to emit a DELETE statement
+for the table itself::
+
+ session.delete(some_b)
+ session.commit()
+
+Will emit SQL as::
+
+ DELETE FROM a WHERE a.id = %(id)s
+ {'id': 1}
+ COMMIT
+
+As always, the target database must have foreign key support with
+ON DELETE CASCADE enabled.
+
+:ticket:`2349`
.. _change_3630:
include_properties=None,
exclude_properties=None,
passive_updates=True,
+ passive_deletes=False,
confirm_deleted_rows=True,
eager_defaults=False,
legacy_is_orphan=False,
ordering for entities. By default mappers have no pre-defined
ordering.
+ :param passive_deletes: Indicates DELETE behavior of foreign key
+ columns when a joined-table inheritance entity is being deleted.
+ Defaults to ``False`` for a base mapper; for an inheriting mapper,
+ defaults to ``False`` unless the value is set to ``True``
+ on the superclass mapper.
+
+ When ``True``, it is assumed that ON DELETE CASCADE is configured
+ on the foreign key relationships that link this mapper's table
+ to its superclass table, so that when the unit of work attempts
+ to delete the entity, it need only emit a DELETE statement for the
+ superclass table, and not this table.
+
+ When ``False``, a DELETE statement is emitted for this mapper's
+ table individually. If the primary key attributes local to this
+ table are unloaded, then a SELECT must be emitted in order to
+ validate these attributes; note that the primary key columns
+ of a joined-table subclass are not part of the "primary key" of
+ the object as a whole.
+
+ Note that a value of ``True`` is **always** forced onto the
+ subclass mappers; that is, it's not possible for a superclass
+ to specify passive_deletes without this taking effect for
+ all subclass mappers.
+
+ .. versionadded:: 1.1
+
+ .. seealso::
+
+ :ref:`passive_deletes` - description of similar feature as
+ used with :func:`.relationship`
+
+ :paramref:`.mapper.passive_updates` - supporting ON UPDATE
+ CASCADE for joined-table inheritance mappers
+
:param passive_updates: Indicates UPDATE behavior of foreign key
columns when a primary key column changes on a joined-table
inheritance mapping. Defaults to ``True``.
:ref:`passive_updates` - description of a similar feature as
used with :func:`.relationship`
+ :paramref:`.mapper.passive_deletes` - supporting ON DELETE
+ CASCADE for joined-table inheritance mappers
+
:param polymorphic_on: Specifies the column, attribute, or
SQL expression used to determine the target class for an
incoming row, when inheriting classes are present.
self._dependency_processors = []
self.validators = util.immutabledict()
self.passive_updates = passive_updates
+ self.passive_deletes = passive_deletes
self.legacy_is_orphan = legacy_is_orphan
self._clause_adapter = None
self._requires_row_aliasing = False
self.inherits._inheriting_mappers.append(self)
self.base_mapper = self.inherits.base_mapper
self.passive_updates = self.inherits.passive_updates
+ self.passive_deletes = self.inherits.passive_deletes or \
+ self.passive_deletes
self._all_tables = self.inherits._all_tables
if self.polymorphic_identity is not None:
(self.polymorphic_identity,
self.polymorphic_map[self.polymorphic_identity],
self, self.polymorphic_identity)
- )
+ )
self.polymorphic_map[self.polymorphic_identity] = self
else:
assert user_roles.count().scalar() == 1
+class PassiveDeletesTest(fixtures.MappedTest):
+ __requires__ = ('foreign_keys',)
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "a", metadata,
+ Column('id', Integer, primary_key=True),
+ Column('type', String(30))
+ )
+ Table(
+ "b", metadata,
+ Column(
+ 'id', Integer, ForeignKey('a.id', ondelete="CASCADE"),
+ primary_key=True),
+ Column('data', String(10))
+ )
+
+ Table(
+ "c", metadata,
+ Column('cid', Integer, primary_key=True),
+ Column('bid', ForeignKey('b.id', ondelete="CASCADE"))
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class A(cls.Basic):
+ pass
+
+ class B(A):
+ pass
+
+ class C(B):
+ pass
+
+ def _fixture(self, a_p=False, b_p=False, c_p=False):
+ A, B, C = self.classes("A", "B", "C")
+ a, b, c = self.tables("a", "b", "c")
+
+ mapper(
+ A, a, passive_deletes=a_p,
+ polymorphic_on=a.c.type, polymorphic_identity='a')
+ mapper(
+ B, b, inherits=A, passive_deletes=b_p, polymorphic_identity='b')
+ mapper(
+ C, c, inherits=B, passive_deletes=c_p, polymorphic_identity='c')
+
+ def test_none(self):
+ A, B, C = self.classes("A", "B", "C")
+ self._fixture()
+
+ s = Session()
+ a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+ s.add_all([a1, b1, c1])
+ s.commit()
+
+ # want to see if the 'C' table loads even though
+ # a and b are loaded
+ c1 = s.query(A).filter_by(id=3).first()
+ s.delete(c1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT c.bid AS c_bid, b.data AS b_data, c.cid AS c_cid "
+ "FROM c, b WHERE :param_1 = b.id AND b.id = c.bid",
+ [{'param_1': 3}]
+ ),
+ CompiledSQL(
+ "DELETE FROM c WHERE c.cid = :cid",
+ [{'cid': 1}]
+ ),
+ CompiledSQL(
+ "DELETE FROM b WHERE b.id = :id",
+ [{'id': 3}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 3}]
+ )
+ )
+
+ def test_c_only(self):
+ A, B, C = self.classes("A", "B", "C")
+ self._fixture(c_p=True)
+
+ s = Session()
+ a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+ s.add_all([a1, b1, c1])
+ s.commit()
+
+ s.delete(a1)
+
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT a.id AS a_id, a.type AS a_type "
+ "FROM a WHERE a.id = :param_1",
+ [{'param_1': 1}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 1}]
+ )
+ )
+
+ b1.id
+ s.delete(b1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM b WHERE b.id = :id",
+ [{'id': 2}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 2}]
+ )
+ )
+
+ # want to see if the 'C' table loads even though
+ # a and b are loaded
+ c1 = s.query(A).filter_by(id=3).first()
+ s.delete(c1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM b WHERE b.id = :id",
+ [{'id': 3}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 3}]
+ )
+ )
+
+ def test_b_only(self):
+ A, B, C = self.classes("A", "B", "C")
+ self._fixture(b_p=True)
+
+ s = Session()
+ a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+ s.add_all([a1, b1, c1])
+ s.commit()
+
+ s.delete(a1)
+
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT a.id AS a_id, a.type AS a_type "
+ "FROM a WHERE a.id = :param_1",
+ [{'param_1': 1}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 1}]
+ )
+ )
+
+ b1.id
+ s.delete(b1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 2}]
+ )
+ )
+
+ c1.id
+ s.delete(c1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 3}]
+ )
+ )
+
+ def test_a_only(self):
+ A, B, C = self.classes("A", "B", "C")
+ self._fixture(a_p=True)
+
+ s = Session()
+ a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+ s.add_all([a1, b1, c1])
+ s.commit()
+
+ s.delete(a1)
+
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT a.id AS a_id, a.type AS a_type "
+ "FROM a WHERE a.id = :param_1",
+ [{'param_1': 1}]
+ ),
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 1}]
+ )
+ )
+
+ b1.id
+ s.delete(b1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 2}]
+ )
+ )
+
+ # want to see if the 'C' table loads even though
+ # a and b are loaded
+ c1 = s.query(A).filter_by(id=3).first()
+ s.delete(c1)
+ with self.sql_execution_asserter(testing.db) as asserter:
+ s.flush()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM a WHERE a.id = :id",
+ [{'id': 3}]
+ )
+ )
+
+
class OptimizedGetOnDeferredTest(fixtures.MappedTest):
"""test that the 'optimized get' path accommodates deferred columns."""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
