(as it is using the sessioncontext plugin, etc), a lazy load operation
will use that session by default if the parent object is not
persistent with a session already.
+- unit-of-work does a better check for "orphaned" objects that are
+part of a "delete-orphan" cascade, for certain conditions where the
+parent isnt available to cascade from.
+- it is now invalid to declare a self-referential relationship with
+"delete-orphan" (as the abovementioned check would make them impossible
+to save)
+- improved the check for objects being part of a session when the
+unit of work seeks to flush() them as part of a relationship..
0.2.7
- quoting facilities set up so that database-specific quoting can be
id=trees.c.node_id,
name=trees.c.node_name,
parent_id=trees.c.parent_node_id,
- children=relation(TreeNode, private=True, backref=backref("parent", foreignkey=trees.c.node_id)),
+ children=relation(TreeNode, cascade="all", backref=backref("parent", foreignkey=trees.c.node_id)),
))
print "\n\n\n----------------------------"
parent_id=trees.c.parent_node_id,
root_id=trees.c.root_node_id,
root=relation(TreeNode, primaryjoin=trees.c.root_node_id==trees.c.node_id, foreignkey=trees.c.node_id, lazy=None, uselist=False),
- children=relation(TreeNode, primaryjoin=trees.c.parent_node_id==trees.c.node_id, lazy=None, uselist=True, cascade="delete,delete-orphan,save-update"),
+ children=relation(TreeNode, primaryjoin=trees.c.parent_node_id==trees.c.node_id, lazy=None, uselist=True, cascade="delete,save-update"),
data=relation(mapper(TreeData, treedata, properties=dict(id=treedata.c.data_id)), cascade="delete,delete-orphan,save-update", lazy=False)
), extension = TreeLoader())
self.properties = properties or {}
self.allow_column_override = allow_column_override
self.allow_null_pks = allow_null_pks
+ self.delete_orphans = []
# a Column which is used during a select operation to retrieve the
# "polymorphic identity" of the row, which indicates which Mapper should be used
# of dependency
#self.compile()
+ def _is_orphan(self, obj):
+ for (key,klass) in self.delete_orphans:
+ if not getattr(klass, key).hasparent(obj):
+ return True
+ else:
+ return False
+
def _get_props(self):
self.compile()
return self.__props
self.target = self.mapper.mapped_table
+ if self.cascade.delete_orphan:
+ if self.parent.class_ is self.mapper.class_:
+ raise exceptions.ArgumentError("Cant establish 'delete-orphan' cascade rule on a self-referential relationship. You probably want cascade='all', which includes delete cascading but not orphan detection.")
+ self.mapper.primary_mapper().delete_orphans.append((self.key, self.parent.class_))
+
if self.secondaryjoin is not None and self.secondary is None:
raise exceptions.ArgumentError("Property '" + self.key + "' specified with secondary join condition but no secondary argument")
# if join conditions were not specified, figure them out based on foreign keys
raise exceptions.InvalidRequestError("Instance '%s' is a detached instance or is already persistent in a different Session" % repr(object))
else:
m = class_mapper(object.__class__, entity_name=kwargs.get('entity_name', None))
+
+ # this would be a nice exception to raise...however this is incompatible with a contextual
+ # session which puts all objects into the session upon construction.
+ #if m._is_orphan(object):
+ # raise exceptions.InvalidRequestError("Instance '%s' is an orphan, and must be attached to a parent object to be saved" % (repr(object)))
+
m._assign_entity_name(object)
self._register_new(object)
pass
def _validate_obj(self, obj):
- if hasattr(obj, '_instance_key') and not self.identity_map.has_key(obj._instance_key):
- raise InvalidRequestError("Instance '%s' is not attached or pending within this session" % repr(obj._instance_key))
-
+ if (hasattr(obj, '_instance_key') and not self.identity_map.has_key(obj._instance_key)) or \
+ (not hasattr(obj, '_instance_key') and obj not in self.new):
+ raise InvalidRequestError("Instance '%s' is not attached or pending within this session" % repr(obj))
+
def update(self, obj):
"""called to add an object to this UnitOfWork as though it were loaded from the DB,
but is actually coming from somewhere else, like a web session or similar."""
continue
if obj in self.deleted:
continue
- flush_context.register_object(obj)
+ if object_mapper(obj)._is_orphan(obj):
+ flush_context.register_object(obj, isdelete=True)
+ else:
+ flush_context.register_object(obj)
for obj in self.deleted:
if objset is not None and not obj in objset:
'orm.sessioncontext',
'orm.objectstore',
+ 'orm.session',
'orm.cascade',
'orm.relationships',
'orm.association',
class C1(Tester):
pass
m1 = mapper(C1, t1, properties = {
- 'c1s':relation(C1, private=True),
+ 'c1s':relation(C1, cascade="all"),
'parent':relation(C1, primaryjoin=t1.c.parent_c1==t1.c.c1, foreignkey=t1.c.c1, lazy=True, uselist=False)
})
a = C1('head c1')
pass
m1 = mapper(C1, t1, properties = {
- 'c1s' : relation(C1, private=True),
+ 'c1s' : relation(C1, cascade="all"),
'c2s' : relation(mapper(C2, t2), private=True)
})
--- /dev/null
+from testbase import AssertMixin
+import testbase
+import unittest, sys, datetime
+
+import tables
+from tables import *
+
+db = testbase.db
+from sqlalchemy import *
+
+
+class SessionTest(AssertMixin):
+
+ def setUpAll(self):
+ db.echo = False
+ tables.create()
+ tables.data()
+ db.echo = testbase.echo
+ def tearDownAll(self):
+ db.echo = False
+ tables.drop()
+ db.echo = testbase.echo
+ def tearDown(self):
+ clear_mappers()
+ def setUp(self):
+ pass
+
+ def test_orphan(self):
+ mapper(Address, addresses)
+ mapper(User, users, properties=dict(
+ addresses=relation(Address, cascade="all,delete-orphan", backref="user")
+ ))
+ s = create_session()
+ a = Address()
+ try:
+ s.save(a)
+ except exceptions.InvalidRequestError, e:
+ pass
+ s.flush()
+ assert a.address_id is None, "Error: address should not be persistent"
+
+ def test_delete_new_object(self):
+ mapper(Address, addresses)
+ mapper(User, users, properties=dict(
+ addresses=relation(Address, cascade="all,delete-orphan", backref="user")
+ ))
+ s = create_session()
+
+ u = User()
+ s.save(u)
+ a = Address()
+ assert a not in s.new
+ u.addresses.append(a)
+ u.addresses.remove(a)
+ s.delete(u)
+ s.flush() # (erroneously) causes "a" to be persisted
+ assert u.user_id is None, "Error: user should not be persistent"
+ assert a.address_id is None, "Error: address should not be persistent"
+
+
+if __name__ == "__main__":
+ testbase.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
