Clone wiki

sqlalchemy / UsageRecipes / PolymorphicOnAnotherTable

PolymorphicOnAnotherTable

This is supported directly by applying a column_property() as the target of polymorphic_on:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class AType(Base):
    __tablename__ = 'atype'

    id = Column(Integer, primary_key=True)
    name = Column(String)

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, primary_key=True)
    data = Column(String)
    type_id = Column(ForeignKey('atype.id'))
    type_name = column_property(select([AType.name]).where(AType.id == id))
    type = relationship(AType)

    __mapper_args__ = {
        "polymorphic_on": type_name,
        "polymorphic_identity": "a"
    }

class ASub(A):
    __tablename__ = 'asub'

    id = Column(ForeignKey('a.id'), primary_key=True)
    __mapper_args__ = {
        "polymorphic_identity": "asub"
    }

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

sess = Session(e)

a_type, asub_type = AType(name="a"), AType(name="asub")

sess.add_all([
    A(data='a1', type=a_type),
    ASub(data='asub1', type=asub_type),
    ASub(data='asub2', type=asub_type),
    A(data='a2', type=a_type),
])
sess.commit()

sess = Session(e)
for a in sess.query(A):
    print a.data, a.type

ALTERNATE METHOD ONE: Populate mapper.polymorphic_map up front with all the polymorphic identities:

# ...

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, primary_key=True)
    data = Column(String)
    type_id = Column(ForeignKey('atype.id'))
    type = relationship(AType)

    __mapper_args__ = {
        "polymorphic_on": type_id,
        "polymorphic_identity": "a"
    }


# ... make sure AType is populated

A.__mapper__.polymorphic_map.update(
    (key, A.__mapper__.polymorphic_map[value])
    for (key, value) in sess.query(AType.id, AType.name)
)

ALTERNATE METHOD TWO: Try to use a Query event to add a JOIN. This recipe needs LOTS OF WORK to be generalized:

# ...

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, primary_key=True)
    data = Column(String)
    type_id = Column(ForeignKey('atype.id'))
    type_name = column_property(AType.name)
    type = relationship(AType)

    __mapper_args__ = {
        "polymorphic_on": type_name,
        "polymorphic_identity": "a"
    }


from sqlalchemy import event
from sqlalchemy.orm.query import Query

@event.listens_for(Query, "before_compile", retval=True)
def setup_query_for_poly(query):
    entities = set(d['entity'] for d in query.column_descriptions)
    if A in entities:
        query = query.join(AType)
    return query

Updated