python - SQLAlchemy ORM: herencia de tabla única polimórfica, con devolución a clase primaria si no se encuentra "identidad polimórfica"

single-table-inheritance

Question

Usando Python 3.5 y SQLAlchemy 1.0.14 (ORM).

Tengo una tabla de artículos declarados como tales:

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

Mis elementos pueden ser de muchos tipos diferentes, el tipo de identificador se almacena en el type . Para algunos de esos tipos de objetos, necesito tener métodos o atributos específicos disponibles.

Para lograr eso, traté de usar la herencia de tabla única con varios SpecialisedItem como subclase de Item:

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"

Ahora, cuando cargue mis artículos, me gustaría que todos los artículos especializados (con el type==''specialitem'' ) se carguen como tales, mientras que cualquier otro tipo de valor daría como resultado el Item clase padre que se está cargando. Eso no funciona, obtengo AssertionError: No such polymorphic_identity ''normal'' is defined al cargar los elementos.

Me gustaría evitar la creación de clases heredadas que no hagan nada solo para cubrir todos los valores de type posibles, en lugar de tener el tipo "no asignado" que vuelve al elemento de la clase padre.

¿Hay alguna forma de lograr ese efecto?

Caso de prueba mínimo para referencia:

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)

Gracias,

Guillaume

Answer 1

Una asignación de identificadores de "identidad polimórfica" a instancias de Mapper se almacena en el dict polymorphic_map . Puede crear un mapa polymorphic_map personalizado que devolverá el mapeador de clases padre para las identidades polimórficas indefinidas.

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_):
    class FallbackToParentPolymorphicMap(dict):
        def __missing__(self, key):
            # return parent Item''s mapper for undefined polymorphic_identity
            return mapper

    new_polymorphic_map = FallbackToParentPolymorphicMap()
    new_polymorphic_map.update(mapper.polymorphic_map)
    mapper.polymorphic_map = new_polymorphic_map

    # for prevent ''incompatible polymorphic identity'' warning, not necessarily
    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)