Wie kann ich eine deklarative Beziehung zu mehreren Tabellen in SQLAlchemy (polimorph) erstellen

Question

Wie kann ich deklarative Beziehung zu mehreren Tabellen in SQLAlchemy (polimorphe) erstellen?

Ich habe Journal-Objekt, das Journal-Tabelle in der Datenbank widerspiegelt. Es enthält Typ und ID (als Fremdschlüssel) in der relevanten Tabelle. Was ich brauche, um eine deklarative Klasse zu erstellen, um diese Tabellen zuzuordnen.

Die Struktur der Datenbank kann nicht geändert werden und kann keine zusätzliche Zuordnungstabelle erstellen. Aber ich kenne Identifikator für Gegenstände.

Was ich versuche zu bekommen, können Sie am Ende des Codebeispiels sehen.

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

Dank Mike article gefunden Lösung von Bayer.

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)