SQLAlchemy多对多关联代理：如何实现重复项静默忽略？

问题场景 #

你通过关联代理实现了Asset与Tag的多对多关联，但当重复调用asset.tags.append(tag)添加已存在的标签时，SQLAlchemy会尝试创建重复的AssetTag关联记录，触发数据库唯一约束冲突：

pymysql.err.IntegrityError: (1062, "Duplicate entry '1-31' for key 'PRIMARY'")

解决方案 #

要让asset.tags具备集合的自动去重行为，可通过以下两步实现：

1. 为关联中间类实现相等性判断 #

修改AssetTag类，添加__eq__和__hash__方法，让集合能识别重复的关联记录：

class AssetTag(BaseModel):
    __tablename__ = 'asset_tags'
    asset_id = sa.Column(sa.Integer, sa.ForeignKey("assets.id"), primary_key=True)
    tag_id = sa.Column(sa.Integer, sa.ForeignKey("tags.id"), primary_key=True)

    asset = orm.relationship("Asset", back_populates='asset_tags')
    tag = orm.relationship("Tag", back_populates='asset_tags')

    def __eq__(self, other):
        if not isinstance(other, AssetTag):
            return False
        return self.asset_id == other.asset_id and self.tag_id == other.tag_id

    def __hash__(self):
        return hash((self.asset_id, self.tag_id))

2. 将关联关系的集合类型改为Set #

修改Asset和Tag类中的asset_tags关系，指定collection_class=set（若需要有序集合，可使用sqlalchemy.util.collections.OrderedSet）：

Asset类修改：

class Asset(BaseModel):
    __tablename__ = 'assets'
    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.VARCHAR(255))

    asset_tags = orm.relationship(
            "AssetTag", back_populates='asset', cascade='all, delete-orphan',
            collection_class=set)  # 改为set类型
    tags = ap.association_proxy(
            "asset_tags", "tag",
            creator=lambda tag: AssetTag(tag=tag))

Tag类修改：

class Tag(BaseModel):
    __tablename__ = 'tags'
    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.VARCHAR(255))

    asset_tags = orm.relationship("AssetTag",
                                  back_populates='tag',
                                  cascade='all, delete-orphan',
                                  collection_class=set)  # 改为set类型
    assets = ap.association_proxy("asset_tags", "asset",
                                  creator=lambda asset: AssetTag(asset=asset))

效果验证 #

修改后，重复调用asset.tags.append(tag)时，SQLAlchemy会自动识别已存在的关联，不会创建重复的AssetTag记录，避免触发唯一约束冲突。

注意：若需要保持关联记录的有序性，可替换collection_class=set为collection_class=OrderedSet，需要先导入from sqlalchemy.util.collections import OrderedSet。

内容的提问来源于stack exchange，提问作者Kaia