SQLAlchemy: how to detect/suppress duplicate JOIN clauses?

Question

Consider the following example code (using SQLAlchemy 1.4):

import os
from sqlalchemy import Column, ForeignKey, Integer, String, select
from sqlalchemy.orm import backref, declarative_base, relationship

Base = declarative_base()

class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    type = Column(String)

class Child(Base):
    __tablename__ = "child"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    parent_id = Column(Integer, ForeignKey("parent.id"))
    parent = relationship("Parent", backref=backref("children"))

def select_parent_name(statement):
    return statement.join(Child.parent).add_columns(Parent.name)

def filter_by_parent_name(statement, parent_name):
    return statement.join(Child.parent).where(Parent.name == parent_name)

def build_query():
    statement = select(Child.id)
    if os.getenv("SELECT_PARENT_NAME", True):
        statement = select_parent_name(statement)
    if os.getenv("FILTER_BY_PARENT", True):
        statement = filter_by_parent_name(statement, "foo")
    return statement

if __name__ == "__main__":
    print(str(build_query()))

This produces invalid SQL, with the same JOIN clause represented twice:

SELECT child.id, parent.name 
FROM child JOIN parent ON parent.id = child.parent_id JOIN parent ON parent.id = child.parent_id 
WHERE parent.name = :name_1

If executed, it will result in:

(MySQLdb._exceptions.OperationalError) (1066, "Not unique table/alias: 'parent'")

It's a stripped-down trivial example, but the point I'm trying to demonstrate is that I'm building up an SQL statement by passing it around to different functions, each of which has a different responsibility and might require the addition of a JOIN which could have already been added to the statement.

Is there an easy way to suppress duplicate JOINs like this? Or to inspect the statement to see if the redundant JOIN is already present? Ideally this information would be easily determined from the statement object itself, rather than having to maintain and pass around that state separately.

user_2345870 · Answer

In SQLAlchemy>=1.4 the joined tables can be found in statement._setup_joins:

joined_tables = [joins[0].parent.entity for joins in statement._setup_joins]

For SQLAlchemy<1.4 the joined tables can be found in statement._join_entities:

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

Reference for SQLAlchemy<1.4: Can I inspect a sqlalchemy query object to find the already joined tables?

SQLAlchemy: how to detect/suppress duplicate JOIN clauses?

1 Answers