SQLAlchemy ORM：多态单 Table 继承，如果未找到 "polymorphic_identity"，则回退到父 class

Question

使用 Python 3.5 和 SQLAlchemy 1.0.14 (ORM)。

我有 table 项这样声明的项目：

from sqlalchemy.ext.declarative.api import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True)
    type = Column(String)
    # other non relevant attributes

我的项目可以有许多不同的类型，类型标识符存储在 type 中。对于其中一些对象类型，我需要具有可用的特定方法或属性。

为了实现这一点，我尝试使用单个 table 继承和多个 SpecialisedItem 作为 Item:class 的子class

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True)
    type = Column(String, index=True)
    # other non relevant attributes

    __mapper_args__ = {
        'polymorphic_on': type,
    }

class SpecialisedItem(Base):
    __mapper_args__ = {
        'polymorphic_identity': 'specialitem',
    }

    def specialised_method(self):
        return "I am special"

现在，当我加载我的项目时，我希望所有特殊项目（具有 type=='specialitem'）都这样加载，而任何其他类型的值将导致父级 class Item 正在加载。这不起作用，我在加载项目时得到 AssertionError: No such polymorphic_identity 'normal' is defined。

我想避免创建仅覆盖所有可能的 type 值的继承 classes，而不是让 "unmapped" type 回落到父级class Item.

有什么方法可以达到这种效果吗？

供参考的最小测试用例：

from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative.api import declarative_base
from sqlalchemy.orm.session import sessionmaker
from sqlalchemy.sql.schema import Column
from sqlalchemy.sql.sqltypes import Integer, String


Base = declarative_base()

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True)
    type = Column(String, index=True)
    # other non relevant attributes

    __mapper_args__ = {
        'polymorphic_on': type,
    }

class SpecialisedItem(Item):
    __mapper_args__ = {
        'polymorphic_identity': 'special',
    }

    specialAttribute = Column(String)

    def specialised_method(self):
        return "I am special"


engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

session.add(Item(type='normal'))
session.add(Item(type='special'))
session.commit()
# loading only specialized items works
for item in session.query(Item).filter_by(type="special"):
    print(item.specialised_method())

# loading other items fails
for item in session.query(Item):
    print(item.type)

谢谢，

纪尧姆

Answer 1

“多态身份”标识符到 Mapper 实例的映射存储在 polymorphic_map 字典中。您可以创建自定义 polymorphic_map，它将 return 父级 class 映射器用于未定义的多态身份。

from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative.api import declarative_base
from sqlalchemy.orm.session import sessionmaker
from sqlalchemy.sql.schema import Column
from sqlalchemy.sql.sqltypes import Integer, String
from sqlalchemy import event

Base = declarative_base()

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True)
    type = Column(String, index=True)
    # other non relevant attributes

    __mapper_args__ = {
        'polymorphic_on': type,
    }

class SpecialisedItem(Item):
    __mapper_args__ = {
        'polymorphic_identity': 'special',
    }

    specialAttribute = Column(String)

    def specialised_method(self):
        return "I am special"

#http://docs.sqlalchemy.org/en/rel_1_1/orm/events.html#sqlalchemy.orm.events.MapperEvents.mapper_configured
@event.listens_for(Item, 'mapper_configured')
def receive_mapper_configured(mapper, class_):    
    mapper.polymorphic_map = defaultdict(lambda: mapper, mapper.polymorphic_map)

    # to prevent 'incompatible polymorphic identity' warning, not mandatory
    mapper._validate_polymorphic_identity = None

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

session.add(Item(type='normal'))
session.add(Item(type='special'))
session.commit()
# loading only specialized items works
for item in session.query(Item).filter_by(type="special"):
    print(item.specialised_method())

# loading other items fails
for item in session.query(Item):
    print(item.type)

Answer 2

基于 @r-m-n 答案的可重用装饰器解决方案。自定义 class 也被替换为实际上做同样事情的 collections.defaultdict。

def receive_mapper_configured(mapper, class_):
    mapper.polymorphic_map = defaultdict(lambda: mapper, mapper.polymorphic_map)
    # to prevent 'incompatible polymorphic identity' warning, not necessary
    mapper._validate_polymorphic_identity = None


def polymorphic_fallback(mapper_klass):
    event.listens_for(mapper_klass, 'mapper_configured')(receive_mapper_configured)
    return mapper_klass

然后在您的代码中，您可以将这个装饰器添加到基础 classes:

@polymorphic_fallback
class Item:
    ...

class SpecificItem(Item):
    ...

SQLAlchemy ORM：多态单 Table 继承，如果未找到 "polymorphic_identity"，则回退到父 class

SQLAlchemy ORM: Polymorphic Single Table Inheritance, with fallback to parent class if "polymorphic_identity" is not found

python

sqlalchemy

single-table-inheritance