How to refer to a column that is the result of an UNION in sqlalchemy?

Question

I have two tables that I'm extracting names from

animal_names = session.query(Animal.name)
human_names = session.query(Human.name)

I want the union of these two

all_names = animal_names.union(human_names)

So that I can join it with another table

query = all_names.join(NameDirectory, NameDirectory.name == <SOMETHING>)

How do I express the right side of the join condition since I don't have a column name for the union?

Looking online I was hoping this would work:

animal_names = session.query(Animal.name).label("name")
human_names = session.query(Human.name).label("name")

all_names = animal_names.union(human_names)
query = all_names.join(NameDirectory, NameDirectory.name == all_names.c.name)

But I get AttributeError: 'Query' object has no attribute 'c'. Is there a way to reference the first column of a Query object to use it in conditions like for JOIN or or WHERE?

Answer 1

I think this might be a problem better solved by SQLAlchemy 1.4/2.0 select() style queries. I was able to rearrange the join and use .subquery() but I think it might be better to use select(). There is a discussion from 2020 about the issues with unions that seems like it might still apply.

Code

from sqlalchemy import (
    create_engine,
    Integer,
    String,
)
from sqlalchemy.schema import (
    Column,
)
from sqlalchemy.sql import select, union
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session


Base = declarative_base()


engine = create_engine("sqlite://", echo=False)


class Animal(Base):
    __tablename__ = 'animals'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)

class Human(Base):
    __tablename__ = 'humans'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)

class NameDirectory(Base):
    __tablename__ = 'name_directories'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)


Base.metadata.create_all(engine)

with Session(engine) as session:
    dog = Animal(name='Spot')
    human = Human(name='Guy')
    human2 = Human(name='Man')
    session.add_all([dog, human, human2])
    for obj in (dog, human):
        session.add(NameDirectory(name=obj.name))
    session.commit()

with Session(engine) as session:
    all_names = session.query(Animal.name.label("name")).union(session.query(Human.name.label("name"))).subquery('all_names')
    query = session.query(NameDirectory.name).join(all_names, all_names.c.name == NameDirectory.name)
    print(query)
    print(query.all())

with Session(engine) as session:
    q1 = select([Animal.name.label('name')]).select_from(Animal)
    q2 = select([Human.name.label('name')]).select_from(Human)
    sub = union(q1, q2).alias()
    q = select([sub.c.name]).select_from(sub.join(NameDirectory, sub.c.name == NameDirectory.name))
    print(q)
    print(session.execute(q).all())

Output

SELECT name_directories.name AS name_directories_name 
FROM name_directories JOIN (SELECT anon_1.name AS name 
FROM (SELECT animals.name AS name 
FROM animals UNION SELECT humans.name AS name 
FROM humans) AS anon_1) AS all_names ON all_names.name = name_directories.name
[('Spot',), ('Guy',)]
SELECT anon_1.name 
FROM (SELECT animals.name AS name 
FROM animals UNION SELECT humans.name AS name 
FROM humans) AS anon_1 JOIN name_directories ON anon_1.name = name_directories.name
[('Spot',), ('Guy',)]