- can use None as a value for cascade.
- documented cascade options in docstring, [ticket:1064]
"0.4.7".
- orm
- - Cascade rules can now function unidirectionally on an
- otherwise bidirectional relation(), taking only
- the cascade behavior of the operation's initiator into
- account. This means that a cascade of "none" on one
- side won't trigger a "save-update" operation when an element
- is attached to that relation, even if the backref
- does indicate "save-update" cascade.
-
+ - The 'cascade' parameter to relation() accepts None
+ as a value, which is equivalent to no cascades.
+
- Added a new SessionExtension hook called after_attach().
This is called at the point of attachment for objects
via add(), add_all(), delete(), and merge().
configurability.
cascade
- a string list of cascade rules which determines how persistence
- operations should be "cascaded" from parent to child.
+ a comma-separated list of cascade rules which determines how Session
+ operations should be "cascaded" from parent to child. This defaults
+ to "False", which means the default cascade should be used.
+ The default value is "save-update, merge".
+ Available cascades are:
+ save-update - cascade the "add()" operation
+ (formerly known as save() and update())
+ merge - cascade the "merge()" operation
+ expunge - cascade the "expunge()" operation
+ delete - cascade the "delete()" operation
+ delete-orphan - if an item of the child's type with no parent is detected,
+ mark it for deletion. Note that this option prevents a pending item
+ of the child's class from being persisted without a parent
+ present.
+ refresh-expire - cascade the expire() and refresh() operations
+ all - shorthand for "save-update,merge, refresh-expire, expunge, delete"
collection_class
a class or function that returns a new list-holding object. will be
return PropertyLoader(argument, secondary=secondary, **kwargs)
def dynamic_loader(argument, secondary=None, primaryjoin=None, secondaryjoin=None, entity_name=None,
- foreign_keys=None, backref=None, post_update=False, cascade=None, remote_side=None, enable_typechecks=True,
+ foreign_keys=None, backref=None, post_update=False, cascade=False, remote_side=None, enable_typechecks=True,
passive_deletes=False, order_by=None):
"""Construct a dynamically-loading mapper property.
backref=None,
_is_backref=False,
post_update=False,
- cascade=None,
+ cascade=False,
viewonly=False, lazy=True,
collection_class=None, passive_deletes=False,
passive_updates=True, remote_side=None,
self._reverse_property = None
- if cascade is not None:
+ if cascade is not False:
self.cascade = CascadeOptions(cascade)
else:
self.cascade = CascadeOptions("save-update, merge")
def __init__(self, key):
self.key = key
- def _target_mapper(self, state):
- prop = _state_mapper(state).get_property(self.key)
- return prop.mapper
-
def append(self, state, item, initiator):
# process "save_update" cascade rules for when an instance is appended to the list of another instance
sess = _state_session(state)
if sess:
- initiating_property = initiator.class_manager[initiator.key].property
- if initiating_property.cascade.save_update and item not in sess:
- sess.save_or_update(item, entity_name=self._target_mapper(state).entity_name)
+ prop = _state_mapper(state).get_property(self.key)
+ if prop.cascade.save_update and item not in sess:
+ sess.save_or_update(item, entity_name=prop.mapper.entity_name)
def remove(self, state, item, initiator):
sess = _state_session(state)
if sess:
- initiating_property = initiator.class_manager[initiator.key].property
+ prop = _state_mapper(state).get_property(self.key)
# expunge pending orphans
- if initiating_property.cascade.delete_orphan and item in sess.new:
- if self._target_mapper(state)._is_orphan(attributes.instance_state(item)):
+ if prop.cascade.delete_orphan and \
+ item in sess.new and \
+ prop.mapper._is_orphan(attributes.instance_state(item)):
sess.expunge(item)
def set(self, state, newvalue, oldvalue, initiator):
return
sess = _state_session(state)
if sess:
- initiating_property = initiator.class_manager[initiator.key].property
- if newvalue is not None and initiating_property.cascade.save_update and newvalue not in sess:
- sess.save_or_update(newvalue, entity_name=self._target_mapper(state).entity_name)
- if initiating_property.cascade.delete_orphan and oldvalue in sess.new:
+ prop = _state_mapper(state).get_property(self.key)
+ if newvalue is not None and prop.cascade.save_update and newvalue not in sess:
+ sess.save_or_update(newvalue, entity_name=prop.mapper.entity_name)
+ if prop.cascade.delete_orphan and oldvalue in sess.new:
sess.expunge(oldvalue)
"""Keeps track of the options sent to relation().cascade"""
def __init__(self, arg=""):
- values = set(c.strip() for c in arg.split(','))
+ if not arg:
+ values = set()
+ else:
+ values = set(c.strip() for c in arg.split(','))
self.delete_orphan = "delete-orphan" in values
self.delete = "delete" in values or "all" in values
self.save_update = "save-update" in values or "all" in values
import testenv; testenv.configure_for_tests()
from testlib.sa import Table, Column, Integer, String, ForeignKey, Sequence
-from testlib.sa.orm import mapper, relation, create_session, class_mapper
+from testlib.sa.orm import mapper, relation, create_session, class_mapper, backref
from testlib.sa.orm import attributes, exc as orm_exc
from testlib import testing
from testlib.testing import eq_
class NoSaveCascadeTest(_fixtures.FixtureTest):
"""test that backrefs don't force save-update cascades to occur
- when they're not desired in the forwards direction."""
+ when the cascade initiated from the forwards side."""
@testing.resolve_artifact_names
def test_unidirectional_cascade_o2m(self):
mapper(Order, orders)
mapper(User, users, properties = dict(
orders = relation(
- Order, cascade="none", backref="user")
+ Order, backref=backref("user", cascade=None))
))
sess = create_session()
assert o1 in sess
sess.clear()
- u1 = User()
- sess.add(u1)
+
o1 = Order()
- o1.user = u1
- assert u1 in sess
+ u1 = User(orders=[o1])
+ sess.add(o1)
+ assert u1 not in sess
assert o1 in sess
@testing.resolve_artifact_names
def test_unidirectional_cascade_m2o(self):
mapper(Order, orders, properties={
- 'user':relation(User, cascade="none", backref="orders")
+ 'user':relation(User, backref=backref("orders", cascade=None))
})
mapper(User, users)
sess = create_session()
-
- o1 = Order()
- sess.add(o1)
- o1.user = u1 = User()
- assert u1 not in sess
- assert o1 in sess
-
- sess.clear()
-
+
+ u1 = User()
+ sess.add(u1)
o1 = Order()
- sess.add(o1)
- u1 = User(orders=[o1])
+ o1.user = u1
+ assert o1 not in sess
assert u1 in sess
- assert o1 in sess
- sess.clear()
-
- o1 = Order()
- o1.user = u1 = User()
- sess.add(o1)
- assert u1 not in sess
- assert o1 in sess
-
sess.clear()
+ u1 = User()
o1 = Order()
- u1 = User(orders=[o1])
+ o1.user = u1
sess.add(u1)
+ assert o1 not in sess
assert u1 in sess
- assert o1 in sess
@testing.resolve_artifact_names
def test_unidirectional_cascade_m2m(self):
sess.add(i1)
k1.items.append(i1)
assert i1 in sess
- assert k1 in sess
+ assert k1 not in sess
class O2MCascadeNoOrphanTest(_fixtures.FixtureTest):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
