SQLAlchemy - 自定义声明性基类中的自引用多对多关系

Question

我正在构建一个使用 SQLAlchemy 的 Python 应用程序，并且我正在尝试在自定义声明性基类与其自身之间实现多对多关系（自引用）。但我无法让它工作。我在下面附上代码以及错误回溯，以防有人可以提供帮助:) 模型的所有实体都已经从这个基类扩展，并且应用程序到目前为止正在运行，以防万一。

谢谢！！

---

代码（非功能性）：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from sqlalchemy import MetaData, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy import String


permissions = Table(
    'permissions', MetaData(),
    Column('origin_id', String, ForeignKey('bases.id'), primary_key=True),
    Column('target_id', String, ForeignKey('bases.id'), primary_key=True)
)


class Base:

    __tablename__ = 'bases'
    __table_args__ = {
        'mysql_engine': 'InnoDB'
    }

    id = Column(String, primary_key=True)

    targets = relationship(
        'Base',
        secondary='permissions',
        primaryjoin='Base.id == permissions.c.origin_id',
        secondaryjoin='Base.id == permissions.c.target_id',
        backref='origins'

        # Reference:
        # https://docs.sqlalchemy.org/en/14/orm/join_conditions.html#self-referential-many-to-many
    )


Base = declarative_base(cls=Base)

追溯：

    class ContactMethod(Base):
  File "/usr/local/lib/python3.8/site-packages/SQLAlchemy-1.4.23-py3.8-linux-x86_64.egg/sqlalchemy/orm/decl_api.py", line 72, in __init__
    _as_declarative(reg, cls, dict_)
  File "/usr/local/lib/python3.8/site-packages/SQLAlchemy-1.4.23-py3.8-linux-x86_64.egg/sqlalchemy/orm/decl_base.py", line 126, in _as_declarative
    return _MapperConfig.setup_mapping(registry, cls, dict_, None, {})
  File "/usr/local/lib/python3.8/site-packages/SQLAlchemy-1.4.23-py3.8-linux-x86_64.egg/sqlalchemy/orm/decl_base.py", line 177, in setup_mapping
    return cfg_cls(registry, cls_, dict_, table, mapper_kw)
  File "/usr/local/lib/python3.8/site-packages/SQLAlchemy-1.4.23-py3.8-linux-x86_64.egg/sqlalchemy/orm/decl_base.py", line 299, in __init__
    self._scan_attributes()
  File "/usr/local/lib/python3.8/site-packages/SQLAlchemy-1.4.23-py3.8-linux-x86_64.egg/sqlalchemy/orm/decl_base.py", line 511, in _scan_attributes
    raise exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Mapper properties (i.e. deferred,column_property(), relationship(), etc.) must be declared as @declared_attr callables on declarative mixin classes.  For dataclass field() objects, use a lambda:

Answer 1

以下是错误输出建议修复的示例：

from sqlalchemy.ext.declarative import declared_attr

class Base:
    ...
    @declared_attr
    def targets(cls):
        return relationship(
            'Base',
            secondary='permissions',
            primaryjoin='Base.id == permissions.c.origin_id',
            secondaryjoin='Base.id == permissions.c.target_id',
            backref='origins'
        )
    ...

旁注：您可以使用基类上的as_declarative mixin 作为快捷方式。

参考文献

扩充基础：https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/mixins.html#augmenting-the-base

declared_attr: https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/api.html#sqlalchemy.ext.declarative.declared_attr

as_declarative：https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/api.html#sqlalchemy.ext.declarative.as_declarative

Answer 2

查看一个有效的自包含示例：

## Imports
from sqlalchemy import Column, ForeignKey, Integer, String, Table, create_engine
from sqlalchemy.orm import Session, as_declarative, declared_attr, relationship

## Configuration
engine = create_engine("sqlite:///:memory:", echo=True)


@as_declarative()
class Base(object):
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


permissions = Table(
    "permissions",
    Base.metadata,
    Column("origin_id", String, ForeignKey("model_base.id"), primary_key=True),
    Column("target_id", String, ForeignKey("model_base.id"), primary_key=True),
)


## Model definitions
class ModelBase(Base):
    __tablename__ = "model_base"
    id = Column(Integer, primary_key=True)
    type = Column(String, nullable=False)

    __mapper_args__ = {
        "polymorphic_on": type,
        "polymorphic_identity": None,
    }

    targets = relationship(
        "ModelBase",
        secondary="permissions",
        primaryjoin="model_base.c.id == permissions.c.origin_id",
        secondaryjoin="model_base.c.id == permissions.c.target_id",
        backref="origins",
        # lazy="raise",
    )


class ClassA(ModelBase):
    __tablename__ = "class_a"
    __mapper_args__ = {"polymorphic_identity": "class_a"}
    id = Column(ForeignKey("model_base.id"), primary_key=True)

    name = Column(String)


class ClassB(ModelBase):
    __tablename__ = "class_b"
    __mapper_args__ = {"polymorphic_identity": "class_b"}
    id = Column(ForeignKey("model_base.id"), primary_key=True)

    name = Column(String)
    value = Column(Integer)


## Tests
def _main():
    with Session(engine) as session:
        Base.metadata.drop_all(engine)
        Base.metadata.create_all(engine)
        print("=" * 80)

        # data
        a1, a2 = [ClassA(name="a1"), ClassA(name="a2")]
        b1, b2 = [ClassB(name="b1"), ClassB(name="b2", value=3)]
        session.add_all([a1, a2, b1, b2])
        session.flush()

        a1.targets.append(a2)
        a1.targets.append(b1)
        a1.targets.append(b2)
        print(b2.targets)
        print(b2.origins)

        session.commit()
        session.expunge_all()

        print("=" * 80)
        a1 = session.query(ClassA).first()
        print(a1)
        print(a1.origins)
        print(a1.targets)

        session.commit()


if __name__ == "__main__":
    _main()