SQLAlchemy - 在具有自定义主连接关系的查询中使用“别名”答案

【问题标题】：SQLAlchemy - Using 'aliased' in query with custom primaryjoin relationshipSQLAlchemy - 在具有自定义主连接关系的查询中使用“别名”
【发布时间】：2014-06-05 14:24:53
【问题描述】：

我在我的 Flask 应用程序中使用 SQLAlchemy (0.9.4)。应用中有两个表支持软删除。

class A(SoftDeleteMixin, db.Model):
    id = db.Column(db.BigInteger, primary_key=True)

    b_id = db.Column(db.BigInteger, db.ForeignKey('b.id'), nullable=False)
    b = soft_delete_relationship('B.id', 'A.b_id')

class B(SoftDeleteMixin, db.Model):
    id = db.Column(db.BigInteger, primary_key=True)
    parent_id = db.Column(db.BigInteger, db.ForeignKey('b.id'))

    parent = soft_delete_relationship(remote(id), parent_id, 'B.id', 'B.parent_id')
    children = soft_delete_relationship(remote(parent_id), id, 'B.parent_id', 'B.id')

SoftDeleteMixin 基于 LimitingQuery (https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PreFilteredQuery)

from sqlalchemy.orm.query import Query


class NonDeletedQuery(Query):
    def get(self, ident):
        return Query.get(self.populate_existing(), ident)

    def __iter__(self):
        return Query.__iter__(self.private())

    def from_self(self, *ent):
        return Query.from_self(self.private(), *ent)

    def private(self):
        mzero = self._mapper_zero()
        if mzero is not None and hasattr(mzero, 'class_'):
            soft_deleted = getattr(mzero.class_, 'soft_deleted', None)
            return self.enable_assertions(False).filter(soft_deleted.is_(False)) if soft_deleted else self
        else:
            return self

并且 soft_delete_relationship 构造与自定义 primaryjoin 的关系（用于非 soft_deleted 上的连接）。

def soft_delete_relationship(first, second, *args, **kwargs):
    if isinstance(first, str) and isinstance(second, str):
        other, other_column = first.split('.')
        _this, this_column = second.split('.')

        primaryjoin = ' & '.join(['({} == {})'.format(first, second), '{}.soft_deleted.is_(False)'.format(other)])

    else:
        other, other_column = args[0].split('.')
        _this, this_column = args[1].split('.')

        primaryjoin = lambda: (first == second) & getattr(second.table.c, 'soft_deleted').is_(False)

    kwargs['primaryjoin'] = primaryjoin
    return relationship(other, **kwargs)

当我使用别名 B 编写查询时出现问题：

b_parent = aliased(B)
A.query.join(A.b).outerjoin(b_parent, B.parent)

我得到以下 SQL：

SELECT ... FROM a JOIN b ON b.id = a.b_id LEFT OUTER JOIN b AS b_1 ON b_1.id = b.parent_id AND *b*.soft_deleted IS False

但我希望如下：

SELECT ... FROM a JOIN b ON b.id = a.b_id LEFT OUTER JOIN b AS b_1 ON b_1.id = b.parent_id AND *b_1*.soft_deleted IS False

当我明确写：

A.query.join(A.b).outerjoin(b_parent, (b_parent.id == B.parent_id) & b_parent.soft_deleted.is_(False))

我得到了正确的查询。

如何在没有显式连接条件的情况下在查询中获得 b_1 的正确别名？顺便说一句，SQLAlchemy 0.7.9 中有预期的 SQL。

【问题讨论】：

标签： python sqlalchemy relationship

【解决方案1】：

好的，我想通了。

getattr(second.table.c, 'soft_deleted') 也必须带有 remote() 注释。

换句话说，B.parent 中的 relationship 中的 primaryjoin 应该如下所示：

(remote(B.id) == B.parent_id) & remote(B.soft_deleted).is_(False)

【讨论】：