test query sort related sqlalchemy

sprox/sa/provider.py

@@ -544,12 +544,7 @@ def query(self, entity, limit=None, offset=None, limit_fields=None,
  for sort_by, sort_descending in zip_longest(order_by, desc):
  if self.is_relation(entity, sort_by):
  mapper = class_mapper(entity)
- class_ = None
- for prop in mapper.iterate_properties:
- try:
- class_ = prop.mapper.class_
- except (AttributeError, KeyError):
- pass
+ class_ = self._get_related_class(entity, sort_by)
  query = query.outerjoin(sort_by)
  f = self.get_view_field_name(class_, related_field_names)
  field = self.get_field(class_, f)

sprox/test/model.py

@@ -158,6 +158,9 @@ class DocumentCategory(DeclarativeBase):
  department_id = Column(Integer, ForeignKey('department.department_id'), primary_key=True)
  name = Column(Unicode(255))
 
+ department = relation(Department, foreign_keys=[department_id])
+
+
 document_category_tag_association_table = Table(
  'document_category_tag_assignment', metadata,
  Column('document_category_id', None, ForeignKey('document_category.document_category_id',

tests/test_saormprovider.py

@@ -252,6 +252,37 @@ def test_query_order_by_multiple(self):
  categories = [(x.owner, x.document_category_id) for x in r]
  assert categories == list(sorted(categories)), (categories, list(sorted(categories)))
 
+ def test_query_sort_related(self):
+ cnt, r = self.provider.query(DocumentCategory, order_by=['department'], related_field_names=['name'])
+ eq_([t.department.name for t in r], ['Accounting', 'Marketing', 'Marketing'])
+
+ cnt, r = self.provider.query(DocumentCategory, order_by=['department'],
+ related_field_names=['name'], desc=True)
+ eq_([t.department.name for t in r], ['Marketing', 'Marketing', 'Accounting'])
+
+ def test_query_sort_related_many(self):
+ g1 = self.provider.create(Group, {'group_name': 'Alpha'})
+ g2 = self.provider.create(Group, {'group_name': 'Beta'})
+ g3 = self.provider.create(Group, {'group_name': 'Zeta'})
+
+ u1 = self.provider.create(User, {'user_name': 'User3', 'groups':[g1]})
+ u2 = self.provider.create(User, {'user_name': 'User1', 'groups':[g2]})
+ u3 = self.provider.create(User, {'user_name': 'User2', 'groups':[g3]})
+
+ cnt, r = self.provider.query(User, order_by=['groups'], related_field_names=['group_name'])
+ eq_([u.user_name for u in r], ['User3', 'User1', 'asdf', 'User2'])
+
+ cnt, r = self.provider.query(User, order_by=['groups'], 
+ related_field_names=['group_name'], desc=True)
+ eq_([u.user_name for u in r], ['User2', 'asdf', 'User1', 'User3'])
+
+ u4 = self.provider.create(User, {'user_name': 'User0', 'groups':[g2, g1]})
+ cnt, r = self.provider.query(User, order_by=['groups'], related_field_names=['group_name'])
+ eq_([u.user_name for u in r], ['User3', 'User0','User1', 'asdf', 'User2'])
+ cnt, r = self.provider.query(User, order_by=['groups'], 
+ related_field_names=['group_name'], desc=True)
+ eq_([u.user_name for u in r], ['User2', 'asdf', 'User1', 'User0', 'User3'])
+
  def test_query_filters(self):
  cnt, r = self.provider.query(Town, filters={'name':'Golden'})
  eq_([t.name for t in r], ['Golden'])