joined_table.py

"""
test multiple layers sqlalchemy inheritance
e.g. Person -> Employee -> Engineer

"""

from sqlalchemy import MetaData, Column, ForeignKey, types, orm
from sqlalchemy.ext.declarative import declarative_base


metadata = MetaData('sqlite://')
metadata.bind.echo = True
BaseClass = declarative_base(metadata=metadata)


class Person(BaseClass):
    __tablename__ = 'person'
    id = Column(types.Integer, primary_key=True)
    name = Column(types.String(20))
    discriminator = Column(types.String(50))
    __mapper_args__ = {'polymorphic_on': discriminator}


class Employee(Person):
    __tablename__ = 'employee'
    id = Column(None, ForeignKey('person.id'), primary_key=True)
    salary = Column(types.Integer)
    __mapper_args__ = {'polymorphic_identity': 'employee'}


class Engineer(Employee):
    __tablename__ = 'engineer'
    id = Column(None, ForeignKey('employee.id'), primary_key=True)
    skill = Column(types.String(50))
    __mapper_args__ = {'polymorphic_identity': 'engineer'}


metadata.create_all()

p = Engineer(name='bob',
             salary=40000,
             skill='java')

sess = orm.create_session()
sess.add(p)
sess.flush()

sess.expunge_all()

testp = sess.query(Person).filter_by(name='bob').one()
print testp.name
print testp.skill
print testp.salary

single_table.py

"""
test multiple layers sqlalchemy inheritance
e.g. Person -> Employee -> Engineer

"""

from sqlalchemy import MetaData, Column, types, orm
from sqlalchemy.ext.declarative import declarative_base


metadata = MetaData('sqlite://')
BaseClass = declarative_base(metadata=metadata)


class Person(BaseClass):
    __tablename__ = 'person'
    id = Column(types.Integer, primary_key=True)
    name = Column(types.String(20))
    discriminator = Column(types.String(50))
    __mapper_args__ = {'polymorphic_on': discriminator}


class Employee(Person):
    __mapper_args__ = {'polymorphic_identity': 'employee'}


class Engineer(Employee):
    __mapper_args__ = {'polymorphic_identity': 'engineer'}


metadata.create_all()

p = Engineer(name='bob')

sess = orm.create_session()
sess.add(p)
sess.flush()

sess.expunge_all()

testp = sess.query(Person).filter_by(name='bob').one()
print testp