Can I inspect a sqlalchemy query object to find the already joined tables?

user_482425 0
Viewed 4740

I'm trying to programmatically build a search query, and to do so, I'm joining a table.

class User(db.Model):
    id = db.Column(db.Integer(), primary_key=True)

class Tag(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    title = db.Column(db.String(128))
    description = db.Column(db.String(128))

This is a bit of a contrived example - I hope it makes sense.

Say my search function looks something like:

def search(title_arg, desc_arg):
    query = User.query
    if title_arg:
        query = query.join(Tag)
        query = query.filter(Tag.title.contains(title_arg))
    if desc_arg:
        query = query.join(Tag)
        query = query.filter(Tag.description.contains(desc_arg))

    return query

Previously, I’ve kept track of what tables that have already been joined in a list, and if the table is in the list, assume it’s already joined, and just add the filter.

It would be cool if I could look at the query object, see that Tag is already joined, and skip it if so. I have some more complex query building that would really benefit from this.

If there’s a completely different strategy for query building for searches that I’ve missed, that would be great too. Or, if the above code is fine if I join the table twice, that's great info as well. Any help is incredibly appreciated!!!

flask-sqlalchemy sqlalchemy join python
3 Answers

Since SQLAlchemy 1.4, the earlier proposed solutions including _join_entities don't work anymore.

SQLAlchemy 1.4

I tried to solve this in SQLAlchemy 1.4, but there is a caveat:

  • This approach includes all entities in the query, so not only joined entities
from sqlalchemy.sql import visitors
from contextlib import suppress

def _has_entity(self, model) -> bool:
    for visitor in visitors.iterate(self.statement):
        # Checking for `.join(Parent.child)` clauses
        if visitor.__visit_name__ == 'binary':
            for vis in visitors.iterate(visitor):
                # Visitor might not have table attribute
                with suppress(AttributeError):
                    # Verify if already present based on table name
                    if model.__table__.fullname == vis.table.fullname:
                        return True
        # Checking for `.join(Child)` clauses
        if visitor.__visit_name__ == 'table':
            # Visitor might be of ColumnCollection or so, 
            # which cannot be compared to model
            with suppress(TypeError):
                if model == visitor.entity_namespace:
                    return True
        # Checking for `Model.column` clauses
        if visitor.__visit_name__ == "column":
            with suppress(AttributeError):
                if model.__table__.fullname == visitor.table.fullname:
                    return True
    return False

def unique_join(self, model, *args, **kwargs):
    """Join if given model not yet in query"""
    if not self._has_entity(model):
        self = self.join(model, *args, **kwargs)
    return self

Query._has_entity = _has_entity
Query.unique_join = unique_join

SQLAlchemy <= 1.3

For SQLAlchemy 1.3 and before, @mtoloo and @r-m-n had perfect answers, I've included them for the sake of completeness.

Some where in your initialization of your project, add a unique_join method to the sqlalchemy.orm.Query object like this:

def unique_join(self, *props, **kwargs):
    if props[0] in [c.entity for c in self._join_entities]:
        return self
    return self.join(*props, **kwargs)

Now use query.unique_join instead of query.join:

Query.unique_join = unique_join
user_85927030
Related