SQLAlchemy（ポリモーフィック）の複数のテーブルに対して宣言的関係を作成する方法

Question

SQLAlchemy（ポリモーフィック）の複数のテーブルに対して宣言的関係を作成するにはどうすればよいですか？

データベースのジャーナルテーブルを反映するJournalオブジェクトがあります。関連するテーブルに型とid（外部キーとして）を保持します。このテーブルをマップする宣言クラスを作成するために必要なもの。

データベースiの構造は変更できません。追加の関連テーブルを作成することはできません。しかし、私はオブジェクトの識別子を知っています。

コードサンプルの一番下に表示されています。

engine = create_engine('sqlite://', echo=True) 
metadata = MetaData(bind=engine) 
Base = declarative_base(bind=metadata) 
DBSession = sessionmaker(bind=engine, autocommit=False, expire_on_commit=False) 
session = DBSession() 


class Person(Base): 
    __abstract__ = True 


class Manager(Person): 
    __tablename__ = 'managers' 

    identifier = '01' 

    id = Column(Integer, primary_key=True) 
    name = Column(String(50)) 
    manager_data = Column(String(40)) 

    __mapper_args__ = { 
     'polymorphic_identity': identifier, 
    } 


class Engineer(Person): 
    __tablename__ = 'engineers' 

    identifier = '02' 

    id = Column(Integer, primary_key=True) 
    name = Column(String(50)) 
    engineer_info = Column(String(40)) 

    __mapper_args__ = { 
     'polymorphic_identity': identifier, 
    } 


class Journal(Base): 
    __tablename__ = 'journal' 

    identifier = '03' 

    id = Column(Integer, primary_key=True) 
    date = Column(Date) 

    type = Column(String(50)) 
    person_id = Column(Integer) 
    person = relationship()  # can’t figure out this relationship 

    __mapper_args__ = { 
     'polymorphic_on': type, 
     'with_polymorphic': '*' 
    } 


if __name__ == '__main__': 

    metadata.create_all() 

    en1 = Engineer(id=1) 
    mn1 = Manager(id=2) 

    session.add_all([en1, mn1]) 
    session.commit() 

    j1 = Journal(person=en1) 
    # -> INSERT INTO Journal (type, person_id) VALUES (’02’, 1) 

    j2 = Journal(person=mn1) 
    # -> INSERT INTO Journal (type, person_id) VALUES (‘01’, 2) 

    for row in session.query(Journal): 
      print(row, row.person) 
    # -> [<Journal …>, <Manager …>] 
    # -> [<Journal …>, <Engineer …>] 

    for row in session.query(Journal).filter(Journal.type == Manager.identifier): 
      print(row, row.person) 
    # -> [<Journal …>, <Manager …>] 

Answer 1

マイクバイヤーのおかげでarticleが見つかりました。

Base = declarative_base() 

class User(Base): 
    __tablename__ = 'users' 
    identifier = '01' 

    id = Column('id', Integer, primary_key=True) 
    name = Column('name', String(50), nullable=False) 

class Order(Base): 
    __tablename__ = 'orders' 
    identifier = '02' 

    id = Column('id', Integer, primary_key=True) 
    description = Column('description', String(50), nullable=False) 


class Address(Base): 
    __tablename__ = 'addresses' 

    id = Column('id', Integer, primary_key=True) 
    addressable_id = Column('addressable_id', Integer) 
    addressable_type = Column('addressable_type', String(50)) 
    street = Column('street', String(100)) 
    city = Column('city', String(50)) 
    country = Column('country', String(50)) 

    def __init__(self, type): 
     self.addressable_type = type 

    member = property(lambda self: getattr(self, '_backref_%s' % self.addressable_type)) 


def addressable(cls, name, uselist=True): 
    """addressable 'interface'.""" 

    def create_address(self): 
     a = Address(cls.identifier) 
     if uselist: 
      getattr(self, name).append(a) 
     else: 
      setattr(self, name, a) 
     return a 

    cls.create_address = create_address 

    mapper = class_mapper(cls) 
    table = mapper.local_table 

    # no constraints. therefore define constraints in an ad-hoc fashion. 
    primaryjoin = and_(
     list(table.primary_key)[0] == Address.addressable_id, 
     Address.addressable_type == cls.identifier 
    ) 

    foreign_keys = [Address.addressable_id] 

    mapper.add_property(name, relation(
     Address, 
     primaryjoin=primaryjoin, uselist=uselist, foreign_keys=foreign_keys, 
     backref=backref(
      '_backref_%s' % cls.identifier, 
      primaryjoin=list(table.primary_key)[0] == Address.addressable_id, 
      foreign_keys=foreign_keys, 
      # lazy="joined" # Joined strategy 
     ) 
    ) 
         ) 


addressable(User, 'addresses', uselist=True) 
addressable(Order, 'address', uselist=False) 

e = create_engine("sqlite://", echo=True) 
Base.metadata.create_all(e) 

u1 = User() 
u1.name = 'bob' 

o1 = Order() 
o1.description = 'order 1' 

a1 = u1.create_address() 
a1.street = '123 anywhere street' 
a2 = u1.create_address() 
a2.street = '345 orchard ave' 

a3 = o1.create_address() 
a3.street = '444 park ave.' 

sess = create_session(bind=e) 
sess.add(u1) 
sess.add(o1) 
sess.flush() 

# query addresses and get relative objects 

for address in sess.query(Address): 
    print("Street", address.street, "Member", address.member)