SQLAlchemy how to join a table from an "aliased" table

user_2946773 0
Viewed 217

I have already read similar questions in SO and on Google, as well as the official SQLAlchemy docs, but still couldn't figure out how to solve my problem.

Consider the following structure (non-relevant fields removed for simplicity):

header_table = Table(
    'header',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('parent_header_id', Integer)
)

item_table = Table(
    'item',
    dal.metadata,
    Column('id', Integer, primary_key=True),
    Column('header_id', Integer)
)

class Header:
    id: int
    parent_header_id: int
    # Relationships
    items: List[Item]
    children: List[Header]

class Item:
    id: int
    header_id: int

mapper(Header, header_table, properties={
    'children': relationship(Header, foreign_keys=[header_table.c.parent_header_id]),
})

Just to summarise: you can nest headers (max of 1 level of nesting), and each header can have items.

I'm trying to load all headers, with their items and children, and the items of the children.

header_alias = aliased(Header)

records = (
    session.query(Header)
        .outerjoin(Header.items)
        .outerjoin(Header.children.of_type(header_alias))
        # .outerjoin(Header.children.of_type(header_alias).items) <<< THE PROBLEM IS HERE (READ BELOW)
        .options(contains_eager(Header.items))
        .options(contains_eager(Header.children.of_type(header_alias)))
        .all()
)

How do I load the items of the children?

The code commented out in the example is wrong, I just put it there as an example of what I'm trying to do.

Note: The code above works, but it's lazy loading the items of the children, I'm trying to get rid of this lazy loading.

sqlalchemy python
1 Answers

Big thanks to @zzzeek (Mike Bayer), author of SQLAlchemy, who answered the question in Github.

https://github.com/sqlalchemy/sqlalchemy/discussions/6876

OK you have to alias "items" also, this is SQL so every table has to be in the FROM clause only once. Here's a full running example

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import aliased
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

Base = declarative_base()

metadata = Base.metadata

header_table = Table(
    "header",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("parent_header_id", ForeignKey("header.id")),
)

item_table = Table(
    "item",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("header_id", ForeignKey("header.id")),
)


class Header(Base):
    __table__ = header_table
    children = relationship("Header")
    items = relationship("Item")


class Item(Base):
    __table__ = item_table
    id: int
    header_id: int


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

s.add(
    Header(
        items=[Item(), Item()],
        children=[Header(items=[Item()]), Header(items=[Item(), Item()])],
    )
)
s.commit()
s.close()


header_alias = aliased(Header)
item_alias = aliased(Item)

records = (
    s.query(Header)
    .outerjoin(Header.items)
    .outerjoin(Header.children.of_type(header_alias))
    .outerjoin(header_alias.items.of_type(item_alias))
    .options(
        contains_eager(Header.items),
        contains_eager(Header.children.of_type(header_alias)).options(
            contains_eager(header_alias.items.of_type(item_alias))
        ),
    )
    .all()
)
s.close()

for r in records:
    print(r)
    print(r.items)
    for c in r.children:
        print(c)
        print(c.items)
user_29467730
Related