Search code examples
pythoninheritancesqlalchemymany-to-manydeclarative

SQLAlchemy polymorphic many-to-many relation with abstract base in declarative syntax


I am creating a SQLAlchemy declarative model with many-to-many relation where one of the tables is concrete and another is polymorphic with abstract base.

Use case: I am sending messages which contain multiple components (component1, component2) and each component has very different set of attributes (so each of the components has own database table). A component can be sent in multiple different messages.

I would like to have a M:N relation between the Message class and an abstract parent class called Component - but the database shouldn't contain any abstract "component" table, only tables for concrete children ("component1", "component2" etc.).

I tried to combine abstract 1:N relation using AbstractConcreteBase (see the last/3rd code snippet in this chapter) with regular M:N relation into the following code which fails to find the name of table for abstract base class:

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base, AbstractConcreteBase

BaseModel = declarative_base()

association = Table("message_component_association", BaseModel.metadata,
    Column("message_id", Integer, ForeignKey("message.id")),
    Column("component_id", Integer, ForeignKey("component.id")))  # Fails to reference the Component class

class Message(BaseModel):
    __tablename__ = "message"
    id = Column(Integer, primary_key=True)
    components = relationship("Component", secondary=association, back_populates="messages")

class Component(AbstractConcreteBase, BaseModel):
    __mapper_args__ = dict(polymorphic_identity="component", concrete=False)  # This seems to be ignored

class Component1(Component):
    __tablename__ = "component1"
    __mapper_args__ = dict(polymorphic_identity="component1", concrete=True)
    id = Column(Integer, primary_key=True)
    messages = relationship("Message", secondary=association, back_populates="components")

engine = create_engine("sqlite://")
BaseModel.metadata.create_all(engine)

The exception says:

sqlalchemy.exc.NoReferencedTableError: Foreign key associated with column 'message_component_association.component_id' could not find table 'component' with which to generate a foreign key to target column 'id'

Why are the mapper_args of Component class ignored and the class cannot be found via provided polymorphic_identity?

EDIT: I've realized I can use Joined Table Inheritance (grr, I cannot post more than 2 links) which replaces the declarative helper mixin by explicit discriminator to get a polymorphic M:N relation - but still, the base class needs own database table.

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

BaseModel = declarative_base()

association = Table("message_component_association", BaseModel.metadata,
Column("message_id", Integer, ForeignKey("message.id")),
Column("component_id", Integer, ForeignKey("component.id")))

class Message(BaseModel):
    __tablename__ = "message"
    id = Column(Integer, primary_key=True)
    components = relationship("Component", secondary=association, back_populates="messages")

class Component(BaseModel):  # Declarative mixin removed
    __tablename__ = "component"  # Requires a real DB table despite being abstract
    __mapper_args__ = dict(polymorphic_identity="component", polymorphic_on="type")  # Apply the discriminator
    id = Column(Integer, primary_key=True)
    type = Column(String(32))  # Explicit discriminator
    messages = relationship("Message", secondary=association, back_populates="components")

class Component1(Component):
    __tablename__ = "component1"
    __mapper_args__ = dict(polymorphic_identity="component1")
    id = Column(Integer, ForeignKey("component.id"), primary_key=True)  # Shares the primary key sequence with the parent and with all other child classes
    messages = relationship("Message", secondary=association, back_populates="components")

engine = create_engine("sqlite://", echo=True)
BaseModel.metadata.create_all(engine)
session = Session(engine)

component_1 = Component1(id=1)
session.commit()

The code seems to work so far but it complains about issues with flush. Is it safe to ignore the warning as long as I don't manually write to the "components table" -- or is there a better way?

SAWarning: Warning: relationship 'messages' on mapper 'Mapper|Component1|component1' supersedes the same relationship on inherited mapper 'Mapper|Component|component'; this can cause dependency issues during flush

Solution

  • SOLUTION: Remove all relationships except the one at Message class and replace back_populates with backref. Backref will create the opposite direction on the fly and the mapper will not see overridden relationships. Also, the polymorphic_identity on the abstract ancestor is not necessary.

    from sqlalchemy import *
    from sqlalchemy.orm import *
    from sqlalchemy.ext.declarative import declarative_base
    
    BaseModel = declarative_base()
    
    association = Table("message_component_association", BaseModel.metadata,
    Column("message_id", Integer, ForeignKey("message.id")),
    Column("component_id", Integer, ForeignKey("component.id")))
    
    class Message(BaseModel):
        __tablename__ = "message"
        id = Column(Integer, primary_key=True)
        components = relationship("Component", secondary=association, backref="messages")  # backref instead of back_populates
    
    class Component(BaseModel):
        __tablename__ = "component"
        __mapper_args__ = dict(polymorphic_on="type")  # Polymorphic identity removed
        id = Column(Integer, primary_key=True)
        type = Column(String(32))
    
        # relationship removed
    
    class Component1(Component):
        __tablename__ = "component1"
        __mapper_args__ = dict(polymorphic_identity="component1")
        id = Column(Integer, ForeignKey("component.id"), primary_key=True)
    
        # relationship removed
    
    engine = create_engine("sqlite://", echo=True)
    BaseModel.metadata.create_all(engine)
    session = Session(engine)
    
    component_1 = Component1(id=1)
    session.commit()