1. idank
  2. sqlalchemy

Commits

Mike Bayer  committed 74e83cd

- got rudimental "mapping to multiple tables" functionality cleaned up,
more correctly documented

  • Participants
  • Parent commits e3dad3a
  • Branches default

Comments (0)

Files changed (7)

File CHANGES

View file
 - count() function on selectables now uses table primary key or 
 first column instead of "1" for criterion, also uses label "rowcount"
 instead of "count".  
+- got rudimental "mapping to multiple tables" functionality cleaned up, 
+more correctly documented
 
 0.2.1
 - "pool" argument to create_engine() properly propigates

File doc/build/content/adv_datamapping.txt

View file
     
     # map to it - the identity of an AddressUser object will be 
     # based on (user_id, address_id) since those are the primary keys involved
-    m = mapper(AddressUser, j)
+    m = mapper(AddressUser, j, properties={
+        'user_id':[users_table.c.user_id, addresses_table.c.user_id]
+    })
 
-    A second example:        
+A second example:
+
     {python}
     # many-to-many join on an association table
     j = join(users_table, userkeywords, 
 
     # 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)
+    m = mapper(KeywordUser, j, properties={
+        'user_id':[users_table.c.user_id, userkeywords.c.user_id],
+        'keyword_id':[userkeywords.c.keyword_id, keywords.c.keyword_id]
+    })
+
+In both examples above, "composite" columns were added as properties to the mappers; these are aggregations of multiple columns into one mapper property, which instructs the mapper to keep both of those columns set at the same value.
 
 ### Mapping a Class against Arbitary Selects {@name=selects}
 

File lib/sqlalchemy/orm/mapper.py

View file
                     l = self.pks_by_table[t]
                 except KeyError:
                     l = self.pks_by_table.setdefault(t, util.HashSet(ordered=True))
-                if not len(t.primary_key):
-                    raise exceptions.ArgumentError("Table " + t.name + " has no primary key columns. Specify primary_key argument to mapper.")
                 for k in t.primary_key:
                     l.append(k)
-
+                    
+        if len(self.pks_by_table[self.mapped_table]) == 0:
+            raise exceptions.ArgumentError("Could not assemble any primary key columsn from given tables for table '%s'" % (self.mapped_table.name))
+            
         # make table columns addressable via the mapper
         self.columns = util.OrderedProperties()
         self.c = self.columns
         list."""
         #print "SAVE_OBJ MAPPER", self.class_.__name__, objects
         connection = uow.transaction.connection(self)
-        for table in self.tables:
+        for table in self.tables.sort(reverse=False):
             #print "SAVE_OBJ table ", self.class_.__name__, table.name
             # looping through our set of tables, which are all "real" tables, as opposed
             # to our main table which might be a select statement or something non-writeable
         connection = uow.transaction.connection(self)
         #print "DELETE_OBJ MAPPER", self.class_.__name__, objects
         
-        for table in util.reversed(self.tables):
+        for table in self.tables.sort(reverse=True):
             if not self._has_pks(table):
                 continue
             delete = []

File lib/sqlalchemy/sql_util.py

View file
         self.tables = []
     def add(self, table):
         self.tables.append(table)
-    def sort(self, reverse=False ):
+        if hasattr(self, '_sorted'):
+            del self._sorted
+    def sort(self, reverse=False):
+        try:
+            sorted = self._sorted
+        except AttributeError, e:
+            self._sorted = self._do_sort()
+            return self.sort(reverse=reverse)
+        if reverse:
+            x = sorted[:]
+            x.reverse()
+            return x
+        else:
+            return sorted
+            
+    def _do_sort(self):
         import sqlalchemy.orm.topological
         tuples = []
         class TVisitor(schema.SchemaVisitor):
                 to_sequence( child )
         if head is not None:
             to_sequence( head )
-        if reverse:
-            sequence.reverse()
         return sequence
         
 

File test/mapper.py

View file
     def testunicodeget(self):
         """tests that Query.get properly sets up the type for the bind parameter.  using unicode would normally fail 
         on postgres, mysql and oracle unless it is converted to an encoded string"""
-        table = Table('foo', db, 
+        metadata = BoundMetaData(db)
+        table = Table('foo', metadata, 
             Column('id', Unicode(10), primary_key=True),
             Column('data', Unicode(40)))
         try:

File test/objectstore.py

View file
         """tests a save of an object where each instance spans two tables. also tests
         redefinition of the keynames for the column properties."""
         usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
-        print usersaddresses.corresponding_column(users.c.user_id)
-        print repr(usersaddresses._orig_cols)
         m = mapper(User, usersaddresses, 
             properties = dict(
                 email = addresses.c.email_address, 
         u.email = 'multi@test.org'
 
         ctx.current.flush()
+        id = m.identity(u)
+        print id
 
+        ctx.current.clear()
+        
+        u = m.get(id)
+        assert u.user_name == 'multitester'
+        
         usertable = users.select(users.c.user_id.in_(u.foo_id)).execute().fetchall()
         self.assertEqual(usertable[0].values(), [u.foo_id, 'multitester'])
         addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
         addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
         self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, 'lala@hey.com'])
 
-        u = m.select(users.c.user_id==u.foo_id)[0]
-        self.echo( repr(u.__dict__))
+        ctx.current.clear()
+        u = m.get(id)
+        assert u.user_name == 'imnew'
+        
+    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)
 
+        # 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'
+        ctx.current.flush()
+        print m.instance_key(k)
+        id = (k.user_id, k.keyword_id)
+        ctx.current.clear()
+        k = ctx.current.query(KeywordUser).get(id)
+        assert k.user_name == 'keyworduser'
+        assert k.keyword_name == 'a keyword'
+        
     def testonetoone(self):
         m = mapper(User, users, properties = dict(
             address = relation(mapper(Address, addresses), lazy = True, uselist = False)

File test/tables.py

View file
 import os
 import testbase
 
-__all__ = ['db', 'users', 'addresses', 'orders', 'orderitems', 'keywords', 'itemkeywords',
+__all__ = ['db', 'users', 'addresses', 'orders', 'orderitems', 'keywords', 'itemkeywords', 'userkeywords',
             'User', 'Address', 'Order', 'Item', 'Keyword'
         ]
 
     
 )
 
+userkeywords = Table('userkeywords', metadata, 
+    Column('user_id', INT, ForeignKey("users")),
+    Column('keyword_id', INT, ForeignKey("keywords")),
+)
+
 itemkeywords = Table('itemkeywords', metadata,
     Column('item_id', INT, ForeignKey("items")),
     Column('keyword_id', INT, ForeignKey("keywords")),