- # -*- coding: utf-8 -*-
- from sqlalchemy import Column, String, Integer, ForeignKey, create_engine
- from sqlalchemy.orm import sessionmaker, relation
- from sqlalchemy.ext import declarative
- engine = create_engine('sqlite:///:memory:')
- Session = sessionmaker(bind=engine)
- session = Session()
- Base = declarative.declarative_base()
- class MailAddress(Base):
- __tablename__ = 'mail_addresses'
- id = Column(Integer, primary_key=True)
- addr = Column(String(256), nullable=False)
- user_id = Column(Integer, ForeignKey('users.id'))
- def __init__(self, addr):
- self.addr = addr
- class Person(Base):
- __tablename__ = 'users'
- id = Column(Integer, primary_key=True)
- name = Column(String(64), nullable=False)
- _discriminator = Column(String(32))
- __mapper_args__ = {'polymorphic_on': _discriminator,
- 'polymorphic_identity': 'user'}
- def __init__(self, name):
- self.name = name
- class Engineer(Person):
- __mapper_args__ = {'polymorphic_identity': 'engineer'}
- mail_addresses = relation('MailAddress', backref='engineer')
- def __init__(self, name, addr):
- super(Engineer, self).__init__(name)
- self.mail_addresses.append(MailAddress(addr))
- Base.metadata.create_all(engine)
- john = Person('John Doe')
- jane = Engineer('Jane Doe', 'jane@example.com')
- session.add_all([john, jane])
- session.commit()
- def get_mail_addresses(person):
- if hasattr(person, 'mail_addresses'):
- return [a.addr for a in person.mail_addresses]
- for p in session.query(Person).all():
- print '%s\'s mail addresses: %s' % \
- (p.name, get_mail_addresses(p))