"Does not link from element" error when eager loading a relationship from an aliased table

Viewed 207

When querying a table, I'd like to eager load a set of columns. Prior to 1.3.x, I could do it with the below code, but it now I'm getting:

sqlalchemy.exc.ArgumentError: Attribute "AliasedClass_Blueprint.engineer" does not link from element "mapped class
 Blueprint->blueprints".  Did you mean to use Building.blueprint.of_type(AliasedClass_Blueprint)?

The query in question is setup as follows:

def doQuery():
    building = aliased(Building)
    blueprint = aliased(Blueprint, name="blueprint")
    engineer = aliased(Engineer, name="engineer")

    with sessionScope() as session:
        return session.query(building)\
            .join(blueprint, blueprint.id==building.blueprintId)\
            .outerjoin(engineer, blueprint.engineerId==engineer.id)\
            .options(contains_eager(building.blueprint, alias=blueprint))\
            .options(contains_eager(building.blueprint, alias=blueprint)\
                 # The error shows up here
                .contains_eager(blueprint.engineer, alias=engineer))

The error suggests using building.blueprint.of_type(blueprint), which seems to work properly, but it looks to be accomplishing something similar to what alias=blueprint does. Does of_type(x) replace the alias=x parameter?

Below is a functioning example with both a working function and one that reproduces the error:

from sqlalchemy import create_engine, inspect, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, aliased, contains_eager
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Engineer(Base):
    __tablename__ = "engineers"
    id=Column(Integer, primary_key=True)
    name=Column(String)

    def __repr__(self):
        return self.name

class Blueprint(Base):
    __tablename__ = "blueprints"
    id=Column(Integer, primary_key=True)
    name=Column(String)
    engineerId=Column(Integer, ForeignKey('engineers.id'))
    engineer=relationship(Engineer, foreign_keys=[engineerId], backref="outputBlueprints")
    def __repr__(self):
        return self.name

class Building(Base):
    __tablename__ = "buildings"

    id=Column(Integer, primary_key=True)
    name=Column(String)
    blueprintId=Column(Integer, ForeignKey('blueprints.id'))
    blueprint = relationship(Blueprint, foreign_keys=[blueprintId], remote_side=[Blueprint.id], backref='buildings')
    def __repr__(self):
        return self.name


engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
sessionFactory = sessionmaker(bind=engine, expire_on_commit=False)

@contextmanager
def sessionScope():
    try:
        session = sessionFactory()
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()
    return

def doQueryWorking():
    building = aliased(Building)
    blueprint = aliased(Blueprint, name="blueprint")
    engineer = aliased(Engineer, name="engineer")

    with sessionScope() as session:
        return session.query(building)\
            .join(blueprint, blueprint.id==building.blueprintId)\
            .outerjoin(engineer, blueprint.engineerId==engineer.id)\
            .options(contains_eager(building.blueprint.of_type(blueprint)))\
            .options(contains_eager(building.blueprint.of_type(blueprint))\
                .contains_eager(blueprint.engineer, alias=engineer))

def doQueryError():
    building = aliased(Building)
    blueprint = aliased(Blueprint, name="blueprint")
    engineer = aliased(Engineer, name="engineer")

    with sessionScope() as session:
        return session.query(building)\
            .join(blueprint, blueprint.id==building.blueprintId)\
            .outerjoin(engineer, blueprint.engineerId==engineer.id)\
            .options(contains_eager(building.blueprint, alias=blueprint))\
            .options(contains_eager(building.blueprint, alias=blueprint)\
                .contains_eager(blueprint.engineer, alias=engineer))

buildings = doQueryError()
0 Answers
Related