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

Question

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!!!

Riess

"Find the already joined table" is a thing you need to skip duplicated joins. It's always a good practice to ask your problem not the means to solve it in a way in your mind. For example ask "how to skip joining to a table if it is already joined". – Soap

Answer 1

You can find joined tables in query._join_entities

joined_tables = [mapper.class_ for mapper in query._join_entities]

Horologe

According to @mtoloo, the joined_tables could also be [mapper.entity for mapper in query._join_entities]. – Pigweed

This worked up and until SQLAlchemy 1.3, but the _join_entities doesn't seem to exist in 1.4 anymore. – Neuroglia

@Neuroglia did you fin any workaround for "_join_entities" ? – Copyread

@Omnia69, not completely yet. But I've found a partial solution for it using visitors from sqlalchemy.sql: v.entity_namespace for v in visitors.iterate(query.statement) if v.__visit_name__ == 'table'. This ONLY works if you use the class directly in the join statement. So it doesn't work for .join(Parent.child), but it does for .join(Child). And it includes entities added in the select statement, which is something you might not want. – Neuroglia

@Neuroglia In 1.4 there is a query._legacy_setup_joins attribute containing joined tables. Since its name contains the "legacy" word I'm not sure if it is something that will be removed in the future or not. It solved my problem similar to yours though. – Nelly

Answer 2

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

Neuroglia

Answer 3

According to the r-m-n answer:

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)

Query.unique_join = unique_join

Now use query.unique_join instead of query.join:

query = query.unique_join(Tag)

Soap

Answer 4

You can check for join tables in SQLAlchemy 2.0 through statement._setup_joins. Might be able to in 1.4 too but haven't got that to check.

Sepal

Answer 5

Sqlalchemy 2.0 version

from sqlalchemy.orm import Query
from sqlalchemy.sql import coercions, roles
from sqlalchemy.sql._typing import _JoinTargetArgument, _OnClauseArgument


def idempotent_join(
    query: Query,
    target: _JoinTargetArgument,
    onclause: _OnClauseArgument | None = None,
    *,
    isouter: bool = False,
    full: bool = False,
) -> Query:
    """Idempotent version of Query.join method"""
    join_target = coercions.expect(roles.JoinTargetRole, target, apply_propagate_attrs=query, legacy=True)
    if onclause is not None:
        onclause_element = coercions.expect(roles.OnClauseRole, onclause, legacy=True)
    else:
        onclause_element = None

    setup_entry = (
        join_target,
        onclause_element,
        None,
        {
            "isouter": isouter,
            "full": full,
        },
    )

    if setup_entry not in query._setup_joins:
        query._setup_joins += (setup_entry,)

        query.__dict__.pop("_last_joined_entity", None)

    return query

Usage:

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

    return query

Wouldbe

Answer 6

You can join several time on the same table by using aliased.

Here is how your function would use it :

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

Geesey

This does not answer the OP's question. – Bayles

That is literaly what he says on his first sentence : "'m trying to programmatically build a search query, and to do so, I'm joining a table.". – Geesey