From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Fri, 10 Aug 2007 15:54:10 +0000 (+0000)
Subject: - merged mapper has_pks fix from r3239 0.3 branch
X-Git-Tag: rel_0_4beta1~18
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=8c3607b0410592b11822b599d3fdc8c7f464944a;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

- merged mapper has_pks fix from r3239 0.3 branch
---

diff --git a/CHANGES b/CHANGES
index d1d55cd449..474ed3d4d3 100644
--- 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 807811ba25..070905d357 100644
--- 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 f224cfc11b..526c575c38 100644
--- 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 e0b0a23d33..68348e42f5 100644
--- 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):