.. automodule:: examples.vertical
+.. _examples_inheritance:
+
Inheritance Mapping Recipes
============================
ability to load elements :term:`polymorphically`, meaning that a single query can
return objects of multiple types.
+.. seealso::
+
+ :ref:`examples_inheritance` - complete exampes of joined, single and
+ concrete inheritance
+
.. _joined_inheritance:
Joined Table Inheritance
"""Working examples of single-table, joined-table, and concrete-table
-inheritance as described in :ref:`datamapping_inheritance`.
+inheritance as described in :ref:`inheritance_toplevel`.
.. autosource::
-"""
\ No newline at end of file
+"""
-"""Concrete (table-per-class) inheritance example."""
+"""Concrete-table (table-per-class) inheritance example."""
-from sqlalchemy import create_engine, MetaData, Table, Column, Integer, \
- String
-from sqlalchemy.orm import mapper, sessionmaker, polymorphic_union
+from sqlalchemy import Column, Integer, String, \
+ ForeignKey, create_engine, inspect, or_
+from sqlalchemy.orm import relationship, Session, with_polymorphic
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.ext.declarative import ConcreteBase
-metadata = MetaData()
-managers_table = Table('managers', metadata,
- Column('employee_id', Integer, primary_key=True),
- Column('name', String(50)),
- Column('manager_data', String(40))
-)
+Base = declarative_base()
-engineers_table = Table('engineers', metadata,
- Column('employee_id', Integer, primary_key=True),
- Column('name', String(50)),
- Column('engineer_info', String(40))
-)
-engine = create_engine('sqlite:///', echo=True)
-metadata.create_all(engine)
+class Company(Base):
+ __tablename__ = 'company'
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ employees = relationship(
+ "Person",
+ back_populates='company',
+ cascade='all, delete-orphan')
-class Employee(object):
- def __init__(self, name):
- self.name = name
def __repr__(self):
- return self.__class__.__name__ + " " + self.name
+ return "Company %s" % self.name
-class Manager(Employee):
- def __init__(self, name, manager_data):
- self.name = name
- self.manager_data = manager_data
- def __repr__(self):
- return self.__class__.__name__ + " " + \
- self.name + " " + self.manager_data
-class Engineer(Employee):
- def __init__(self, name, engineer_info):
- self.name = name
- self.engineer_info = engineer_info
+class Person(ConcreteBase, Base):
+ __tablename__ = 'person'
+ id = Column(Integer, primary_key=True)
+ company_id = Column(ForeignKey('company.id'))
+ name = Column(String(50))
+
+ company = relationship("Company", back_populates="employees")
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'person',
+ }
+
def __repr__(self):
- return self.__class__.__name__ + " " + \
- self.name + " " + self.engineer_info
+ return "Ordinary person %s" % self.name
+
+
+class Engineer(Person):
+ __tablename__ = 'engineer'
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ company_id = Column(ForeignKey('company.id'))
+ status = Column(String(30))
+ engineer_name = Column(String(30))
+ primary_language = Column(String(30))
+ company = relationship("Company", back_populates="employees")
-pjoin = polymorphic_union({
- 'manager':managers_table,
- 'engineer':engineers_table
-}, 'type', 'pjoin')
+ __mapper_args__ = {
+ 'polymorphic_identity': 'engineer',
+ 'concrete': True
+ }
-employee_mapper = mapper(Employee, pjoin, polymorphic_on=pjoin.c.type)
-manager_mapper = mapper(Manager, managers_table,
- inherits=employee_mapper, concrete=True,
- polymorphic_identity='manager')
-engineer_mapper = mapper(Engineer, engineers_table,
- inherits=employee_mapper, concrete=True,
- polymorphic_identity='engineer')
+ def __repr__(self):
+ return (
+ "Engineer %s, status %s, engineer_name %s, "
+ "primary_language %s" %
+ (
+ self.name, self.status,
+ self.engineer_name, self.primary_language)
+ )
+
+
+class Manager(Person):
+ __tablename__ = 'manager'
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ company_id = Column(ForeignKey('company.id'))
+ status = Column(String(30))
+ manager_name = Column(String(30))
+
+ company = relationship("Company", back_populates="employees")
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'manager',
+ 'concrete': True
+ }
+
+ def __repr__(self):
+ return "Manager %s, status %s, manager_name %s" % (
+ self.name, self.status, self.manager_name)
+
+
+engine = create_engine('sqlite://', echo=True)
+Base.metadata.create_all(engine)
+
+session = Session(engine)
+
+c = Company(name='company1', employees=[
+ Manager(
+ name='pointy haired boss',
+ status='AAB',
+ manager_name='manager1'),
+ Engineer(
+ name='dilbert',
+ status='BBA',
+ engineer_name='engineer1',
+ primary_language='java'),
+ Person(name='joesmith'),
+ Engineer(
+ name='wally',
+ status='CGG',
+ engineer_name='engineer2',
+ primary_language='python'),
+ Manager(
+ name='jsmith',
+ status='ABA',
+ manager_name='manager2')
+])
+session.add(c)
+
+session.commit()
+c = session.query(Company).get(1)
+for e in c.employees:
+ print(e, inspect(e).key, e.company)
+assert set([e.name for e in c.employees]) == set(
+ ['pointy haired boss', 'dilbert', 'joesmith', 'wally', 'jsmith'])
+print("\n")
-session = sessionmaker(engine)()
+dilbert = session.query(Person).filter_by(name='dilbert').one()
+dilbert2 = session.query(Engineer).filter_by(name='dilbert').one()
+assert dilbert is dilbert2
-m1 = Manager("pointy haired boss", "manager1")
-e1 = Engineer("wally", "engineer1")
-e2 = Engineer("dilbert", "engineer2")
+dilbert.engineer_name = 'hes dilbert!'
-session.add(m1)
-session.add(e1)
-session.add(e2)
session.commit()
-print(session.query(Employee).all())
+c = session.query(Company).get(1)
+for e in c.employees:
+ print(e)
+
+# query using with_polymorphic.
+eng_manager = with_polymorphic(Person, [Engineer, Manager])
+print(
+ session.query(eng_manager).
+ filter(
+ or_(
+ eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2'
+ )
+ ).all()
+)
+
+# illustrate join from Company
+eng_manager = with_polymorphic(Person, [Engineer, Manager])
+print(
+ session.query(Company).
+ join(
+ Company.employees.of_type(eng_manager)
+ ).filter(
+ or_(eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2')
+ ).all())
+session.commit()
"""Joined-table (table-per-subclass) inheritance example."""
-from sqlalchemy import Table, Column, Integer, String, \
+from sqlalchemy import Column, Integer, String, \
ForeignKey, create_engine, inspect, or_
from sqlalchemy.orm import relationship, Session, with_polymorphic
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
+
class Company(Base):
__tablename__ = 'company'
id = Column(Integer, primary_key=True)
name = Column(String(50))
- employees = relationship("Person",
- backref='company',
- cascade='all, delete-orphan')
+ employees = relationship(
+ "Person",
+ back_populates='company',
+ cascade='all, delete-orphan')
def __repr__(self):
return "Company %s" % self.name
+
class Person(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True)
- company_id = Column(Integer, ForeignKey('company.id'))
+ company_id = Column(ForeignKey('company.id'))
name = Column(String(50))
type = Column(String(50))
+ company = relationship("Company", back_populates="employees")
+
__mapper_args__ = {
- 'polymorphic_identity':'person',
- 'polymorphic_on':type
+ 'polymorphic_identity': 'person',
+ 'polymorphic_on': type
}
+
def __repr__(self):
return "Ordinary person %s" % self.name
+
class Engineer(Person):
__tablename__ = 'engineer'
- id = Column(Integer, ForeignKey('person.id'), primary_key=True)
+ id = Column(ForeignKey('person.id'), primary_key=True)
status = Column(String(30))
engineer_name = Column(String(30))
primary_language = Column(String(30))
__mapper_args__ = {
- 'polymorphic_identity':'engineer',
+ 'polymorphic_identity': 'engineer',
}
+
def __repr__(self):
- return "Engineer %s, status %s, engineer_name %s, "\
- "primary_language %s" % \
- (self.name, self.status,
- self.engineer_name, self.primary_language)
+ return (
+ "Engineer %s, status %s, engineer_name %s, "
+ "primary_language %s" %
+ (
+ self.name, self.status,
+ self.engineer_name, self.primary_language)
+ )
+
class Manager(Person):
__tablename__ = 'manager'
- id = Column(Integer, ForeignKey('person.id'), primary_key=True)
+ id = Column(ForeignKey('person.id'), primary_key=True)
status = Column(String(30))
manager_name = Column(String(30))
__mapper_args__ = {
- 'polymorphic_identity':'manager',
+ 'polymorphic_identity': 'manager',
}
+
def __repr__(self):
- return "Manager %s, status %s, manager_name %s" % \
- (self.name, self.status, self.manager_name)
+ return "Manager %s, status %s, manager_name %s" % (
+ self.name, self.status, self.manager_name)
engine = create_engine('sqlite://', echo=True)
name='pointy haired boss',
status='AAB',
manager_name='manager1'),
- Engineer(name='dilbert',
+ Engineer(
+ name='dilbert',
status='BBA',
engineer_name='engineer1',
primary_language='java'),
Person(name='joesmith'),
- Engineer(name='wally',
- status='CGG',
- engineer_name='engineer2',
- primary_language='python'),
- Manager(name='jsmith',
- status='ABA',
- manager_name='manager2')
+ Engineer(
+ name='wally',
+ status='CGG',
+ engineer_name='engineer2',
+ primary_language='python'),
+ Manager(
+ name='jsmith',
+ status='ABA',
+ manager_name='manager2')
])
session.add(c)
c = session.query(Company).get(1)
for e in c.employees:
print(e, inspect(e).key, e.company)
-assert set([e.name for e in c.employees]) == set(['pointy haired boss',
- 'dilbert', 'joesmith', 'wally', 'jsmith'])
+assert set([e.name for e in c.employees]) == set(
+ ['pointy haired boss', 'dilbert', 'joesmith', 'wally', 'jsmith'])
print("\n")
dilbert = session.query(Person).filter_by(name='dilbert').one()
print(e)
# query using with_polymorphic.
-eng_manager = with_polymorphic(Person, [Engineer, Manager], aliased=True)
-print(session.query(eng_manager).\
- filter(
- or_(eng_manager.Engineer.engineer_name=='engineer1',
- eng_manager.Manager.manager_name=='manager2'
- )
- ).all())
-
-# illustrate join from Company,
-# We use aliased=True
-# to help when the selectable is used as the target of a join.
-eng_manager = with_polymorphic(Person, [Engineer, Manager], aliased=True)
-print(session.query(Company).\
+eng_manager = with_polymorphic(Person, [Engineer, Manager])
+print(
+ session.query(eng_manager).
+ filter(
+ or_(
+ eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2'
+ )
+ ).all()
+)
+
+# illustrate join from Company.
+# flat=True means the tables inside the "polymorphic join" will be aliased.
+# not strictly necessary in this example but helpful for the more general
+# case of joins involving inheritance hierarchies as well as joined eager
+# loading.
+eng_manager = with_polymorphic(Person, [Engineer, Manager], flat=True)
+print(
+ session.query(Company).
join(
- eng_manager,
- Company.employees
+ Company.employees.of_type(eng_manager)
).filter(
- or_(eng_manager.Engineer.engineer_name=='engineer1',
- eng_manager.Manager.manager_name=='manager2')
+ or_(eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2')
).all())
session.commit()
-
-
-"""Single-table inheritance example."""
-
-from sqlalchemy import MetaData, Table, Column, Integer, String, \
- ForeignKey, create_engine
-from sqlalchemy.orm import mapper, relationship, sessionmaker
-
-metadata = MetaData()
-
-# a table to store companies
-companies = Table('companies', metadata,
- Column('company_id', Integer, primary_key=True),
- Column('name', String(50)))
-
-employees_table = Table('employees', metadata,
- Column('employee_id', Integer, primary_key=True),
- Column('company_id', Integer, ForeignKey('companies.company_id')),
- Column('name', String(50)),
- Column('type', String(20)),
- Column('status', String(20)),
- Column('engineer_name', String(50)),
- Column('primary_language', String(50)),
- Column('manager_name', String(50))
-)
+"""Single-table (table-per-hierarchy) inheritance example."""
+
+from sqlalchemy import Column, Integer, String, \
+ ForeignKey, create_engine, inspect, or_
+from sqlalchemy.orm import relationship, Session, with_polymorphic
+from sqlalchemy.ext.declarative import declarative_base, declared_attr
+
+Base = declarative_base()
+
+class Company(Base):
+ __tablename__ = 'company'
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+
+ employees = relationship(
+ "Person",
+ back_populates='company',
+ cascade='all, delete-orphan')
+
+ def __repr__(self):
+ return "Company %s" % self.name
+
+
+class Person(Base):
+ __tablename__ = 'person'
+ id = Column(Integer, primary_key=True)
+ company_id = Column(ForeignKey('company.id'))
+ name = Column(String(50))
+ type = Column(String(50))
+
+ company = relationship("Company", back_populates="employees")
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'person',
+ 'polymorphic_on': type
+ }
-class Person(object):
- def __init__(self, **kwargs):
- for key, value in kwargs.items():
- setattr(self, key, value)
def __repr__(self):
return "Ordinary person %s" % self.name
+
+
class Engineer(Person):
- def __repr__(self):
- return "Engineer %s, status %s, engineer_name %s, "\
- "primary_language %s" % \
- (self.name, self.status,
- self.engineer_name, self.primary_language)
-class Manager(Person):
- def __repr__(self):
- return "Manager %s, status %s, manager_name %s" % \
- (self.name, self.status, self.manager_name)
-class Company(object):
- def __init__(self, **kwargs):
- for key, value in kwargs.items():
- setattr(self, key, value)
- def __repr__(self):
- return "Company %s" % self.name
-person_mapper = mapper(Person, employees_table,
- polymorphic_on=employees_table.c.type,
- polymorphic_identity='person')
-manager_mapper = mapper(Manager, inherits=person_mapper,
- polymorphic_identity='manager')
-engineer_mapper = mapper(Engineer, inherits=person_mapper,
- polymorphic_identity='engineer')
+ engineer_name = Column(String(30))
+ primary_language = Column(String(30))
-mapper(Company, companies, properties={
- 'employees': relationship(Person, lazy=True, backref='company')
-})
+ # illustrate a single-inh "conflicting" column declaration;
+ # see http://docs.sqlalchemy.org/en/latest/orm/extensions/
+ # declarative/inheritance.html#resolving-column-conflicts
+ @declared_attr
+ def status(cls):
+ return Person.__table__.c.get('status', Column(String(30)))
+ __mapper_args__ = {
+ 'polymorphic_identity': 'engineer',
+ }
-engine = create_engine('sqlite:///', echo=True)
+ def __repr__(self):
+ return (
+ "Engineer %s, status %s, engineer_name %s, "
+ "primary_language %s" %
+ (
+ self.name, self.status,
+ self.engineer_name, self.primary_language)
+ )
+
+
+class Manager(Person):
+ manager_name = Column(String(30))
-metadata.create_all(engine)
+ @declared_attr
+ def status(cls):
+ return Person.__table__.c.get('status', Column(String(30)))
-session = sessionmaker(engine)()
+ __mapper_args__ = {
+ 'polymorphic_identity': 'manager',
+ }
-c = Company(name='company1')
-c.employees.append(Manager(name='pointy haired boss', status='AAB',
- manager_name='manager1'))
-c.employees.append(Engineer(name='dilbert', status='BBA',
- engineer_name='engineer1', primary_language='java'))
-c.employees.append(Person(name='joesmith', status='HHH'))
-c.employees.append(Engineer(name='wally', status='CGG',
- engineer_name='engineer2', primary_language='python'
- ))
-c.employees.append(Manager(name='jsmith', status='ABA',
- manager_name='manager2'))
+ def __repr__(self):
+ return "Manager %s, status %s, manager_name %s" % (
+ self.name, self.status, self.manager_name)
+
+
+engine = create_engine('sqlite://', echo=True)
+Base.metadata.create_all(engine)
+
+session = Session(engine)
+
+c = Company(name='company1', employees=[
+ Manager(
+ name='pointy haired boss',
+ status='AAB',
+ manager_name='manager1'),
+ Engineer(
+ name='dilbert',
+ status='BBA',
+ engineer_name='engineer1',
+ primary_language='java'),
+ Person(name='joesmith'),
+ Engineer(
+ name='wally',
+ status='CGG',
+ engineer_name='engineer2',
+ primary_language='python'),
+ Manager(
+ name='jsmith',
+ status='ABA',
+ manager_name='manager2')
+])
session.add(c)
+
session.commit()
c = session.query(Company).get(1)
for e in c.employees:
- print(e, e.company)
-
+ print(e, inspect(e).key, e.company)
+assert set([e.name for e in c.employees]) == set(
+ ['pointy haired boss', 'dilbert', 'joesmith', 'wally', 'jsmith'])
print("\n")
dilbert = session.query(Person).filter_by(name='dilbert').one()
dilbert2 = session.query(Engineer).filter_by(name='dilbert').one()
assert dilbert is dilbert2
-dilbert.engineer_name = 'hes dibert!'
+dilbert.engineer_name = 'hes dilbert!'
-session.flush()
-session.expunge_all()
+session.commit()
c = session.query(Company).get(1)
for e in c.employees:
print(e)
-session.delete(c)
+# query using with_polymorphic.
+eng_manager = with_polymorphic(Person, [Engineer, Manager])
+print(
+ session.query(eng_manager).
+ filter(
+ or_(
+ eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2'
+ )
+ ).all()
+)
+
+# illustrate join from Company,
+eng_manager = with_polymorphic(Person, [Engineer, Manager])
+print(
+ session.query(Company).
+ join(
+ Company.employees.of_type(eng_manager)
+ ).filter(
+ or_(eng_manager.Engineer.engineer_name == 'engineer1',
+ eng_manager.Manager.manager_name == 'manager2')
+ ).all())
+
session.commit()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
