SQLAlchemy (polimorphic)에서 여러 테이블에 선언적 관계를 만들 수있는 방법

Question

SQLAlchemy (polimorphic)에서 여러 테이블에 선언적 관계를 만들려면 어떻게해야합니까?

데이터베이스의 저널 테이블을 반영하는 저널 개체가 있습니다. 관련 테이블에 type과 id (외래 키)를 가지고있다. 이 테이블을 매핑하는 데 선언 클래스를 만드는 데 필요한 것은 무엇입니까.

데이터베이스의 구조는 변경할 수 없으며 추가 연결 테이블을 만들 수 없습니다. 하지만 객체에 대한 식별자를 알고 있습니다.

내가 얻으려고하는 것은 코드 예제의 맨 아래에서 볼 수 있습니다.

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

Mike Bayer의 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)