Clone wiki

sqlalchemy / UsageRecipes / DeclarativeAbstractConcreteBase


Note: This feature will be included as of SQLAlchemy 0.7.3. See].

A mixin class that allows declarative to be used with concrete inheritance, including a base class that queries polymorphically among subclasses using a UNION.

The key to this recipe is to delay the production of mapper() objects until all table metadata has been defined.

from sqlalchemy.orm.util import polymorphic_union
from sqlalchemy.orm import mapper

class DeclarativeAbstractConcreteBase(object):

    _mapper_args = []

    def __mapper_cls__(cls, *args, **kw):
        """Declarative will use this function in lieu of 
        calling mapper() directly.

        Collect each series of arguments and invoke
        them when prepare() is called.

        cls._mapper_args.append((args, kw))

    def prepare(cls):
        """Initialize the mappings and polymorphic union."""

        # create a polymorphic_union as described in the
        # docs.  We generate it based on what we find in 
        # __subclasses__().
        pjoin = polymorphic_union(dict(
            (klass.__mapper_args__['polymorphic_identity'], klass.__table__)
            for klass in cls.__subclasses__()
            if hasattr(klass, '__mapper_args__')
        ), 'type', 'pjoin')

        # create a mapper() on the base.
        # this can be altered to map to cls.__table__, using with_polymporphic
        # as the receiver of 'pjoin' instead, if the base class is to be 
        # concrete as well - "Employee" in our example below would
        # include 'Base' in its declaration.
        cls.__mapper__ = mapper(cls, pjoin, polymorphic_on=pjoin.c.type)

        # iterate through our collected sets of mapper
        # args and call mapper() for each.
        for args, kw in cls._mapper_args:
            klass = args[0]
            # the registry is global to the 
            # base here so ensure we only deal with 
            # classes in this hierarchy 
            if not issubclass(klass, cls):
            if kw.get('concrete', False):
                kw.setdefault('inherits', cls)
            klass.__mapper__ = mapper(*args, **kw)

if __name__ == '__main__':
    from sqlalchemy import create_engine, Column, Integer, \
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class Employee(DeclarativeAbstractConcreteBase):

    class Manager(Base, Employee):
        __tablename__ = 'manager'
        employee_id = Column(Integer, primary_key=True)
        name = Column(String(50))
        manager_data = Column(String(40))
        __mapper_args__ = {

    class Engineer(Base, Employee):
        __tablename__ = 'engineer'
        employee_id = Column(Integer, primary_key=True)
        name = Column(String(50))
        engineer_info = Column(String(40))
        __mapper_args__ = {'polymorphic_identity':'engineer', 

    # setup mappings

    e = create_engine('sqlite://', echo=True)


    session = Session(e)

    m1 = Manager(name="pointy haired boss", manager_data="manager1")
    e1 = Engineer(name="wally", engineer_info="engineer1")
    e2 = Engineer(name="dilbert", engineer_info="engineer2")


    print session.query(Employee).all()