Commits

Mike Bayer committed 1518be1

- convert Query tuples to "named" tuples

  • Participants
  • Parent commits 19d1f0c
  • Branches user_defined_state

Comments (0)

Files changed (4)

lib/sqlalchemy/orm/attributes.py

     def __clause_element__(self):
         return self.comparator.__clause_element__()
     
+    def label(self, name):
+        return self.__clause_element__().label(name)
+        
     def operate(self, op, *other, **kwargs):
         return op(self.comparator, *other, **kwargs)
 

lib/sqlalchemy/orm/query.py

             filter = None
 
         custom_rows = single_entity and 'append_result' in self._entities[0].extension.methods
-        
-        process = [query_entity.row_processor(self, context, custom_rows) for query_entity in self._entities]
 
+        (process, labels) = zip(*[query_entity.row_processor(self, context, custom_rows) for query_entity in self._entities])
+
+        if not single_entity:
+            labels = dict([(label, property(util.itemgetter(i))) for i, label in enumerate(labels) if label])
+            rowtuple = type.__new__(type, "RowTuple", (tuple,), labels)
+            rowtuple.keys = labels.keys
+            
         while True:
             context.progress = util.Set()
             context.partials = {}
             elif single_entity:
                 rows = [process[0](context, row) for row in fetch]
             else:
-                rows = [tuple([proc(context, row) for proc in process]) for row in fetch]
+                rows = [rowtuple([proc(context, row) for proc in process]) for row in fetch]
 
             if filter:
                 rows = filter(rows)
             def main(context, row):
                 return _instance(row, None)
         
-        return main
+        if self.is_aliased_class:
+            entname = self.entity._sa_label_name
+        else:
+            entname = self.mapper.class_.__name__
+            
+        return main, entname
 
     def setup_context(self, query, context):
         # if single-table inheritance mapper, add "typecol IN (polymorphic)" criterion so
 
         def proc(context, row):
             return row[column]
-        return proc
+            
+        return (proc, getattr(column, 'name', None))
 
     def setup_context(self, query, context):
         column = self.__resolve_expr_against_query_aliases(query, self.column, context)

lib/sqlalchemy/orm/util.py

         sql_util.ColumnAdapter.__init__(self, selectable, equivalents, chain_to)
 
 class AliasedClass(object):
-    def __init__(self, cls, alias=None):
+    def __init__(self, cls, alias=None, name=None):
         self.__mapper = _class_to_mapper(cls)
         self.__target = self.__mapper.class_
         alias = alias or self.__mapper._with_polymorphic_selectable.alias()
         self.__adapter = sql_util.ClauseAdapter(alias, equivalents=self.__mapper._equivalent_columns)
         self.__alias = alias
+        self._sa_label_name = name
         self.__name__ = 'AliasedClass_' + str(self.__target)
 
     def __adapt_prop(self, prop):

test/orm/query.py

         # whereas this uses users.c.xxx, is not aliased and creates a new join
         q2 = q.select_from(sel).filter(users.c.id==8).filter(users.c.id>sel.c.id).values(users.c.name, sel.c.name, User.name)
         self.assertEquals(list(q2), [(u'ed', u'jack', u'jack')])
+    
+    def test_tuple_labeling(self):
+        sess = create_session()
+        for row in sess.query(User, Address).join(User.addresses).all():
+            self.assertEquals(set(row.keys()), set(['User', 'Address']))
+            self.assertEquals(row.User, row[0])
+            self.assertEquals(row.Address, row[1])
+            
+        for row in sess.query(User.name, User.id.label('foobar')):
+            self.assertEquals(set(row.keys()), set(['name', 'foobar']))
+            self.assertEquals(row.name, row[0])
+            self.assertEquals(row.foobar, row[1])
+
+        for row in sess.query(User).values(User.name, User.id.label('foobar')):
+            self.assertEquals(set(row.keys()), set(['name', 'foobar']))
+            self.assertEquals(row.name, row[0])
+            self.assertEquals(row.foobar, row[1])
+
+        oalias = aliased(Order)
+        for row in sess.query(User, oalias).join(User.orders).all():
+            self.assertEquals(set(row.keys()), set(['User']))
+            self.assertEquals(row.User, row[0])
+
+        oalias = aliased(Order, name='orders')
+        for row in sess.query(User, oalias).join(User.orders).all():
+            self.assertEquals(set(row.keys()), set(['User', 'orders']))
+            self.assertEquals(row.User, row[0])
+            self.assertEquals(row.orders, row[1])
 
     def test_column_queries(self):
         sess = create_session()