test_joined_inheritance.py

import logging
from os import remove

from sqlalchemy import (create_engine, Column, ForeignKey, Integer, String, Unicode)
from sqlalchemy.ext.declarative import (declarative_base, declared_attr)
from sqlalchemy.orm import (configure_mappers, sessionmaker)
from sqlalchemy.util import classproperty
from sqlalchemy_continuum import make_versioned

logging.basicConfig()
logging.getLogger('sqlalchemy.orm').setLevel(logging.INFO)
make_versioned(user_cls=None)

Base = declarative_base()

class Node(Base):
    __versioned__ = {}
    __mapper_args__ = dict(polymorphic_on='type',
        polymorphic_identity='node',
        with_polymorphic='*')

    @declared_attr
    def __tablename__(cls):
        return '{0}s'.format(cls.__name__.lower())

    id = Column(Integer(), primary_key=True)
    type = Column(String(30), nullable=False)
    name = Column(Unicode)

class Content(Node):
    __versioned__ = {}

    @classproperty
    def __mapper_args__(cls):
        return dict(polymorphic_identity=cls.__name__.lower())

    id = Column(Integer, ForeignKey('nodes.id'), primary_key=True)


class Document(Content):
    __versioned__ = {}
    id = Column(Integer(), ForeignKey('contents.id'), primary_key=True)

if __name__ == '__main__':

    configure_mappers()

    DB_FILE = 'test.db'
    try:
        remove(DB_FILE)
    except OSError:
        pass
    engine = create_engine('sqlite:///{}'.format(DB_FILE))

    Base.metadata.bind = engine
    Base.metadata.create_all()

    session = sessionmaker(bind=engine)()

    d1 = Document(name=u'd1')
    session.add(d1)
    d2 = Document(name=u'd2')
    session.add(d2)
    d3 = Document(name=u'd3')
    session.add(d3)

    session.commit()