SQLAlchemy - Using 'aliased' in query with

问题:

I'm using SQLAlchemy (0.9.4) in my Flask application. There are two tables with soft delete support in application.

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 is based on 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

And soft_delete_relationship constructs relationship with custom primaryjoin (for join on non-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)

The problem occurs when I write query with aliased B:

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

I get following 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

But I expect following:

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

When I explicitly write:

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

I got right query.

How can I get proper alias to b_1 in query without explicit join condition? Btw, there was expected SQL in SQLAlchemy 0.7.9.

回答1:

OK, I figured it out.

getattr(second.table.c, 'soft_deleted') must be also with remote() annotation.

In other words primaryjoin of relationship in B.parent should look like:

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