- merged mapper has_pks fix from r3239 0.3 branch

diff --git a/CHANGES b/CHANGES
index d1d55cd44915beee142f00c0a3c2bbb6edba5533..474ed3d4d3ec981bc5621341496af5cc0d9f2cf7 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -296,6 +296,8 @@
     - added a check for joining from A->B using join(), along two
       different m2m tables.  this raises an error in 0.3 but is 
       possible in 0.4 when aliases are used. [ticket:687]
+    - fixed bug where mapper, being linked to a join where one table had 
+      no PK columns, would not detect that the joined table had no PK.
 - engine
     - fixed another occasional race condition which could occur
       when using pool with threadlocal setting

diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 807811ba25968fe164056c6fc8eaf52dfd5c03d8..070905d357e355d1e5cc87f3c9b6ce176762935c 100644 (file)
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -1292,7 +1292,10 @@ class Mapper(object):
 
     def _has_pks(self, table):
         try:
-            for k in self.pks_by_table[table]:
+            pk = self.pks_by_table[table]
+            if not pk:
+                return False
+            for k in pk:
                 if not self.columntoproperty.has_key(k):
                     return False
             else:

diff --git a/test/orm/mapper.py b/test/orm/mapper.py
index f224cfc11b23194508b8e8cae153ae2270a33828..526c575c3840594e2dd7ef9abf75d2d3951c18db 100644 (file)
--- a/test/orm/mapper.py
+++ b/test/orm/mapper.py
@@ -308,6 +308,30 @@ class MapperTest(MapperSuperTest):
         q = create_session().query(m)
         l = q.select()
         self.assert_result(l, User, *user_result[0:2])
+    
+    def testmappingtojoinnopk(self):
+        metadata = MetaData()
+        account_ids_table = Table('account_ids', metadata,
+                Column('account_id', Integer, primary_key=True),
+                Column('username', String(20)))
+        account_stuff_table = Table('account_stuff', metadata,
+                Column('account_id', Integer, ForeignKey('account_ids.account_id')),
+                Column('credit', Numeric))
+        class A(object):pass
+        m = mapper(A, account_ids_table.join(account_stuff_table))
+        m.compile()
+        assert m._has_pks(account_ids_table)
+        assert not m._has_pks(account_stuff_table)
+        metadata.create_all(testbase.db)
+        try:
+            sess = create_session(bind=testbase.db)
+            a = A()
+            sess.save(a)
+            sess.flush()
+            assert testbase.db.execute(account_ids_table.count()).scalar() == 1
+            assert testbase.db.execute(account_stuff_table.count()).scalar() == 0
+        finally:
+            metadata.drop_all(testbase.db)
         
     def testmappingtoouterjoin(self):
         """test mapping to an outer join, with a composite primary key that allows nulls"""

diff --git a/test/orm/unitofwork.py b/test/orm/unitofwork.py
index e0b0a23d33032c8fc6b3fb70d84ba871b37ad5d4..68348e42f50b7921ac2c711d5e8cef01f6954d08 100644 (file)
--- a/test/orm/unitofwork.py
+++ b/test/orm/unitofwork.py
@@ -1476,36 +1476,6 @@ class ManyToManyTest(UnitOfWorkTest):
         Session.close()
         l = Item.query.filter(items.c.item_name.in_(*[e['item_name'] for e in data[1:]])).order_by(items.c.item_name).all()
         self.assert_result(l, *data)
-
-    def testm2mmultitable(self):
-        # many-to-many join on an association table
-        j = join(users, userkeywords, 
-                users.c.user_id==userkeywords.c.user_id).join(keywords, 
-                   userkeywords.c.keyword_id==keywords.c.keyword_id)
-        print "PK", j.primary_key
-        # a class 
-        class KeywordUser(object):
-            pass
-
-        # map to it - the identity of a KeywordUser object will be
-        # (user_id, keyword_id) since those are the primary keys involved
-        m = mapper(KeywordUser, j, properties={
-            'user_id':[users.c.user_id, userkeywords.c.user_id],
-            'keyword_id':[userkeywords.c.keyword_id, keywords.c.keyword_id],
-            'keyword_name':keywords.c.name,
-        }, )
-
-        k = KeywordUser()
-        k.user_name = 'keyworduser'
-        k.keyword_name = 'a keyword'
-        Session.commit()
-        
-        id = (k.user_id, k.keyword_id)
-        Session.close()
-        k = Session.query(KeywordUser).get(id)
-        assert k.user_name == 'keyworduser'
-        assert k.keyword_name == 'a keyword'
-
     
 class SaveTest2(UnitOfWorkTest):