- many-to-many relationships properly set the type of bind params

for delete operations on the association table
- many-to-many relationships check that the number of rows deleted
from the association table by a delete operation matches the expected
results

diff --git a/CHANGES b/CHANGES
index daa2e1e57ed6e2b8c8b59b14ca7956349d89c214..43c4ab88f554cd5d25c9b5955b1254a4d69f2a59 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -5,6 +5,11 @@
 - orm
     - "delete-orphan" no longer implies "delete". ongoing effort to 
       separate the behavior of these two operations.
+    - many-to-many relationships properly set the type of bind params
+      for delete operations on the association table
+    - many-to-many relationships check that the number of rows deleted
+      from the association table by a delete operation matches the expected 
+      results
 - mysql
     - support for column-level CHARACTER SET and COLLATE declarations,
       as well as ASCII, UNICODE, NATIONAL and BINARY shorthand.

diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index e623fd2a1ddbb5a103bae36cab3fd04f347563c1..0c0dacd2028a325db3be662f013fa55f2fe6b09a 100644 (file)
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -362,11 +362,15 @@ class ManyToManyDP(DependencyProcessor):
                     self._synchronize(obj, child, associationrow, False, uowcommit)
                     uowcommit.attributes[(self, "manytomany", obj, child)] = True
                     secondary_delete.append(associationrow)
+
         if len(secondary_delete):
             secondary_delete.sort()
             # TODO: precompile the delete/insert queries?
-            statement = self.secondary.delete(sql.and_(*[c == sql.bindparam(c.key) for c in self.secondary.c if c.key in associationrow]))
-            connection.execute(statement, secondary_delete)
+            statement = self.secondary.delete(sql.and_(*[c == sql.bindparam(c.key, type=c.type) for c in self.secondary.c if c.key in associationrow]))
+            result = connection.execute(statement, secondary_delete)
+            if result.supports_sane_rowcount() and result.rowcount != len(secondary_delete):
+                raise exceptions.ConcurrentModificationError("Deleted rowcount %d does not match number of objects deleted %d" % (result.rowcount, len(secondary_delete)))
+
         if len(secondary_insert):
             statement = self.secondary.insert()
             connection.execute(statement, secondary_insert)

diff --git a/test/orm/relationships.py b/test/orm/relationships.py
index 6fb666dab6c86cc05fc373a988860a4224bf06b0..5f53af080f5754a2d9de18b712b357637e8d7f0a 100644 (file)
--- a/test/orm/relationships.py
+++ b/test/orm/relationships.py
@@ -673,6 +673,56 @@ class TypeMatchTest(testbase.ORMTest):
         except exceptions.AssertionError, err:
             assert str(err) == "Attribute 'a' on class '%s' doesn't handle objects of type '%s'" % (D, B)
 
+class TypedAssociationTable(testbase.ORMTest):
+    def define_tables(self, metadata):
+        global t1, t2, t3
+        
+        class MySpecialType(TypeDecorator):
+            impl = String
+            def convert_bind_param(self, value, dialect):
+                return "lala" + value
+            def convert_result_value(self, value, dialect):
+                return value[4:]
+            
+        t1 = Table('t1', metadata, 
+            Column('col1', MySpecialType(30), primary_key=True),
+            Column('col2', String(30)))
+        t2 = Table('t2', metadata, 
+            Column('col1', MySpecialType(30), primary_key=True),
+            Column('col2', String(30)))
+        t3 = Table('t3', metadata,
+            Column('t1c1', MySpecialType(30), ForeignKey('t1.col1')),
+            Column('t2c1', MySpecialType(30), ForeignKey('t2.col1')),
+        )
+    def testm2m(self):
+        """test many-to-many tables with special types for candidate keys"""
+        
+        class T1(object):pass
+        class T2(object):pass
+        mapper(T2, t2)
+        mapper(T1, t1, properties={
+            't2s':relation(T2, secondary=t3, backref='t1s')
+        })
+        a = T1()
+        a.col1 = "aid"
+        b = T2()
+        b.col1 = "bid"
+        c = T2()
+        c.col1 = "cid"
+        a.t2s.append(b)
+        a.t2s.append(c)
+        sess = create_session()
+        sess.save(a)
+        sess.flush()
+
+        assert t3.count().scalar() == 2
+        
+        a.t2s.remove(c)
+        sess.flush()
+        
+        assert t3.count().scalar() == 1
+        
+# TODO: move these tests to either attributes.py test or its own module
 class CustomCollectionsTest(testbase.ORMTest):
     def define_tables(self, metadata):
         global sometable, someothertable