<p>改编自:<a href="http://www.duanqu.tech/questions/2898814/sqlalchemy-polymorphic-association" rel="nofollow noreferrer">http://www.duanqu.tech/questions/2898814/sqlalchemy-polymorphic-association</a>
和<a href="http://docs.sqlalchemy.org/en/latest/orm/inheritance.html" rel="nofollow noreferrer">http://docs.sqlalchemy.org/en/latest/orm/inheritance.html</a></p>
<pre><code>from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy import create_engine, Integer, Column, \
String, ForeignKey
from sqlalchemy.orm import Session, relationship
from sqlalchemy_utils import database_exists, create_database
from sqlalchemy.dialects.postgresql import UUID
import uuid
@as_declarative()
class Base(object):
# Set the tablenames
# to the class names
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
# give an id to each class
id = Column(UUID, default=lambda: str(uuid.uuid4()), primary_key=True)
# Base = declarative_base()
class EntityInterface(Base):
discriminator = Column(String)
__mapper_args__ = {"polymorphic_on": discriminator}
class Address(Base):
street = Column(String)
city = Column(String)
zip = Column(String)
entity_id = Column(UUID, ForeignKey(EntityInterface.id),
default=lambda: str(uuid.uuid4()))
entity = relationship(EntityInterface)
def __repr__(self):
return ("%s(street=%r, city=%r, zip=%r, company=%r)" %
(self.__class__.__name__, self.street, self.city,
self.zip, self.entity))
class Person(EntityInterface):
id = Column(UUID, ForeignKey(EntityInterface.id),
default=lambda: str(uuid.uuid4()), primary_key=True)
name = Column(String)
__mapper_args__ = {"polymorphic_identity": "Person"}
class Organization(EntityInterface):
id = Column(UUID, ForeignKey(EntityInterface.id),
default=lambda: str(uuid.uuid4()), primary_key=True)
name = Column(String)
__mapper_args__ = {"polymorphic_identity": "Organization"}
engine = create_engine(
'postgresql://paul_tsc:paul123@localhost/' + "poly_testing",
echo=False)
# if it doesn't exist, create it
if not database_exists(engine.url):
create_database(engine.url)
Base.metadata.create_all(engine)
session = Session(engine)
address1 = Address(street='test-1', city="Detroit", zip="56785")
address2 = Address(street='test-2', city="Phoenix", zip="110322")
address3 = Address(street='test-3', city="Washington", zip="432414")
org1 = Organization(name="Org-1 TEST")
org2 = Organization(name="Org-2 TEST")
person1 = Person(name="Person-1 TEST")
person2 = Person(name="Person-2 TEST")
address1.entity = org1
address2.entity = person1
address3.entity = person1
session.add_all([address1, address2, address3])
session.commit()
address3.entity = org2
session.commit()
print("PRINTING, TOTAL = %s" % session.query(Address).count())
for address in session.query(Address):
print("ADDRESS = %s" % address)
</code></pre>