按 sqlalchemy 中的连接别名排序

Posted 2023-03-24

【中文标题】按 sqlalchemy 中的连接别名排序

【英文标题】：Order by a join alias in sqlalchemy

【发布时间】：2016-12-31 09:08:22

【问题描述】：

出于某种原因，我在 sqlalchemy 中使用了手动联接（Query.join 未加入加载）。我使用了别名，因为我有多个连接到一个表。现在我想按关系字段之一对结果进行排序。如何使用带有别名的 Query.order_by？当我这样做时，我得到一个模棱两可的 %(####) 而不是查询中的字段名称。

        if self.order_by:
        entity = self.cls
        for field, order in self.order_by:

            if '.' in field:

                m = re.match(r'(.+)\.(.+)', field)

                if m.group(1) not in self.aliases:

                    for item in m.group(1).split('.'):

                        cls = inspect(entity)
                        attr = cls.attrs[item]
                        entity = get_type(attr)

                        if attr.innerjoin:
                            aliased_entity = aliased(entity)
                            self.aliases[m.group(1)] = aliased_entity
                            _query = _query.join(aliased_entity, item).options(contains_eager(item,
                                                                                        alias=aliased_entity))
                        else:
                            aliased_entity = aliased(entity)
                            self.aliases[m.group(1)] = aliased_entity
                            _query = _query.outerjoin(aliased_entity, item).options(contains_eager(item,
                                                                                                 alias=aliased_entity))

            if order == "desc":
                _query = _query.order_by(self.get_order_by_field(field).desc())
            else:
                _query = _query.order_by(self.get_order_by_field(field).asc())

然后：

    def get_order_by_field(self, field: str) -> Column:
        if '.' in field:
            m = re.match(r'(.+)\.(.+)', field)
            if m.group(1) in self.aliases:
                return getattr(self.aliases[m.group(1)], m.group(2))
            else:
                return Column(self.column_map[field])
        else:
            return Column(field)

【参考方案1】：

参见示例：

entity = sqlalchemy.aliased(ModelUser)
session.query(ModelLog.id, ModelLog.date, ModelUser.id.label('current_user_id'),
              entity.id.label('prev_user_id'))) \
       .join(ModelUser, ModelUser.id == ModelLog.id_model_user) \
       .join(entity, entity.id == ModelLog.id_prev_model_user) \
       .filter(...) \
       .order_by(entity.id.asc(), ModelUser.id.desc()

【讨论】：

我必须把部分代码放在这里。上面的评论。