the "check for orphans" step will cascade the delete operation to child objects.

diff --git a/lib/sqlalchemy/orm/unitofwork.py b/lib/sqlalchemy/orm/unitofwork.py
index 00d34a10480d425098f46e0ab8989f4975f5fee2..c0bb219a32e1e6056fabefb0f9daa72f1fe8372b 100644 (file)
--- a/lib/sqlalchemy/orm/unitofwork.py
+++ b/lib/sqlalchemy/orm/unitofwork.py
@@ -180,18 +180,24 @@ class UnitOfWork(object):
         else:
             objset = None
 
+        processed = util.Set()
         for obj in [n for n in self.new] + [d for d in self.dirty]:
             if objset is not None and not obj in objset:
                 continue
-            if obj in self.deleted:
+            if obj in self.deleted or obj in processed:
                 continue
             if object_mapper(obj)._is_orphan(obj):
-                flush_context.register_object(obj, isdelete=True)
+                for c in [obj] + list(object_mapper(obj).cascade_iterator('delete', obj)):
+                    if c in processed:
+                        continue
+                    flush_context.register_object(c, isdelete=True)
+                    processed.add(c)
             else:
                 flush_context.register_object(obj)
-            
+                processed.add(obj)
+                
         for obj in self.deleted:
-            if objset is not None and not obj in objset:
+            if (objset is not None and not obj in objset) or obj in processed:
                 continue
             flush_context.register_object(obj, isdelete=True)
         

diff --git a/test/orm/session.py b/test/orm/session.py
index 4bc5264d1558cbedb73d5f514d1ea7479df56bee..2be9ef67f904e663495b6bf7f0a4cacac1bb2f8a 100644 (file)
--- a/test/orm/session.py
+++ b/test/orm/session.py
@@ -9,7 +9,7 @@ db = testbase.db
 from sqlalchemy import *
 
 
-class SessionTest(AssertMixin):
+class OrphanDeletionTest(AssertMixin):
 
     def setUpAll(self):
         db.echo = False
@@ -58,5 +58,66 @@ class SessionTest(AssertMixin):
         assert a.address_id is None, "Error: address should not be persistent"
 
 
+class CascadingOrphanDeletionTest(AssertMixin):
+    def setUpAll(self):
+        global meta, orders, items, attributes
+        meta = BoundMetaData(db)
+
+        orders = Table('orders', meta,
+            Column('id', Integer, Sequence('order_id_seq'), primary_key = True),
+            Column('name', VARCHAR(50)),
+
+        )
+        items = Table('items', meta,
+            Column('id', Integer, Sequence('item_id_seq'), primary_key = True),
+            Column('order_id', Integer, ForeignKey(orders.c.id), nullable=False),
+            Column('name', VARCHAR(50)),
+
+        )
+        attributes = Table('attributes', meta,
+            Column('id', Integer, Sequence('attribute_id_seq'), primary_key = True),
+            Column('item_id', Integer, ForeignKey(items.c.id), nullable=False),
+            Column('name', VARCHAR(50)),
+
+        )
+
+    def setUp(self):
+        meta.create_all()
+    def tearDown(self):
+        meta.drop_all()
+
+    def testdeletechildwithchild(self):
+        class Order(object): pass
+        class Item(object): pass
+        class Attribute(object): pass
+
+        attrMapper = mapper(Attribute, attributes)
+        itemMapper = mapper(Item, items, properties=dict(
+            attributes=relation(attrMapper, cascade="all,delete-orphan", backref="item")
+        ))
+        orderMapper = mapper(Order, orders, properties=dict(
+            items=relation(itemMapper, cascade="all,delete-orphan", backref="order")
+        ))
+
+        s = create_session(echo_uow=True)
+        order = Order()
+        s.save(order)
+
+        item = Item()
+        attr = Attribute()
+        item.attributes.append(attr)
+
+        order.items.append(item)
+        order.items.remove(item) # item is an orphan, but attr is not so flush() tries to save attr
+        s.flush()
+
+        assert item.id is None
+        assert attr.id is None
+
+
+
+
+
+
 if __name__ == "__main__":    
     testbase.main()