Source code for examples.generic_associations.generic_fk
"""Illustrates a so-called "generic foreign key", in a similar fashionto that of popular frameworks such as Django, ROR, etc. Thisapproach bypasses standard referential integritypractices, in that the "foreign key" column is not actuallyconstrained to refer to any particular table; instead,in-application logic is used to determine which table is referenced.This approach is not in line with SQLAlchemy's usual style, as foregoingforeign key integrity means that the tables can easily contain invalidreferences and also have no ability to use in-database cascade functionality.However, due to the popularity of these systems, as well as that it usesthe fewest number of tables (which doesn't really offer any "advantage",though seems to be comforting to many) this recipe remains inhigh demand, so in the interests of having an easy StackOverflow answerqueued up, here it is. The author recommends "table_per_related"or "table_per_association" instead of this approach."""fromsqlalchemyimportand_fromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimporteventfromsqlalchemyimportIntegerfromsqlalchemyimportStringfromsqlalchemy.ext.declarativeimportas_declarativefromsqlalchemy.ext.declarativeimportdeclared_attrfromsqlalchemy.ormimportbackreffromsqlalchemy.ormimportforeignfromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportremotefromsqlalchemy.ormimportSession@as_declarative()classBase:"""Base class which provides automated table name and surrogate primary key column. """@declared_attrdef__tablename__(cls):returncls.__name__.lower()id=Column(Integer,primary_key=True)classAddress(Base):"""The Address class. This represents all address records in a single table. """street=Column(String)city=Column(String)zip=Column(String)discriminator=Column(String)"""Refers to the type of parent."""parent_id=Column(Integer)"""Refers to the primary key of the parent. This could refer to any table. """@propertydefparent(self):"""Provides in-Python access to the "parent" by choosing the appropriate relationship. """returngetattr(self,"parent_%s"%self.discriminator)def__repr__(self):return"%s(street=%r, city=%r, zip=%r)"%(self.__class__.__name__,self.street,self.city,self.zip,)classHasAddresses:"""HasAddresses mixin, creates a relationship to the address_association table for each parent. """@event.listens_for(HasAddresses,"mapper_configured",propagate=True)defsetup_listener(mapper,class_):name=class_.__name__discriminator=name.lower()class_.addresses=relationship(Address,primaryjoin=and_(class_.id==foreign(remote(Address.parent_id)),Address.discriminator==discriminator,),backref=backref("parent_%s"%discriminator,primaryjoin=remote(class_.id)==foreign(Address.parent_id),),)@event.listens_for(class_.addresses,"append")defappend_address(target,value,initiator):value.discriminator=discriminatorclassCustomer(HasAddresses,Base):name=Column(String)classSupplier(HasAddresses,Base):company_name=Column(String)engine=create_engine("sqlite://",echo=True)Base.metadata.create_all(engine)session=Session(engine)session.add_all([Customer(name="customer 1",addresses=[Address(street="123 anywhere street",city="New York",zip="10110"),Address(street="40 main street",city="San Francisco",zip="95732"),],),Supplier(company_name="Ace Hammers",addresses=[Address(street="2569 west elm",city="Detroit",zip="56785")],),])session.commit()forcustomerinsession.query(Customer):foraddressincustomer.addresses:print(address)print(address.parent)