SQLAlchemy多对多关系：UNIQUE约束冲突的最优解决方法问询

针对你遇到的多对多关联中批量添加触发Category.text唯一约束错误的问题，以下是几种无需手动遍历去重的高效解决方案：

方案1：封装get_or_create批量处理关联对象 #

通过一次查询获取已存在的Category，仅创建不存在的项，避免重复插入：

from sqlalchemy.orm import Session

def get_or_create_categories(db: Session, texts: list[str]) -> list[Category]:
    # 查询数据库中已存在的Category
    existing_cats = db.query(Category).filter(Category.text.in_(texts)).all()
    existing_texts = {cat.text for cat in existing_cats}
    
    # 创建不存在的Category
    new_cats = [Category(text=text) for text in texts if text not in existing_texts]
    db.add_all(new_cats)
    db.flush()  # 刷新会话，获取新创建对象的ID
    
    # 构建文本到对象的映射，返回与输入顺序匹配的Category列表
    cat_map = {cat.text: cat for cat in existing_cats + new_cats}
    return [cat_map[text] for text in texts]

使用时，先收集所有视频关联的Category文本，统一处理后再关联到Video：

# 示例视频数据
videos_data = [
    {"title": "视频1", "categories": ["red", "blue"]},
    {"title": "视频2", "categories": ["red", "green"]}
]

# 收集所有需要的Category文本
all_cat_texts = {text for data in videos_data for text in data["categories"]}
# 获取或创建所有Category
categories = get_or_create_categories(db, list(all_cat_texts))
cat_map = {c.text: c for c in categories}

# 创建Video并关联Category
videos = []
for data in videos_data:
    video = Video(title=data["title"])
    video.categories = [cat_map[text] for text in data["categories"]]
    videos.append(video)

db.add_all(videos)
db.commit()

方案2：数据库层面忽略重复插入 #

利用数据库原生语法处理重复项，不同数据库语法略有差异：

MySQL（使用INSERT IGNORE） #

from sqlalchemy import text

# 批量插入Category，忽略重复项
insert_sql = text("""
    INSERT IGNORE INTO category (text)
    VALUES (:text)
""")
db.execute(insert_sql, [{"text": t} for t in all_cat_texts])
db.commit()

PostgreSQL（使用ON CONFLICT DO NOTHING） #

from sqlalchemy import text

insert_sql = text("""
    INSERT INTO category (text)
    VALUES (:text)
    ON CONFLICT (text) DO NOTHING
""")
db.execute(insert_sql, [{"text": t} for t in all_cat_texts])
db.commit()

执行完上述语句后，直接查询所有需要的Category关联到Video即可。

方案3：SQLAlchemy原生批量冲突处理 #

使用SQLAlchemy针对不同数据库的方言扩展，实现批量插入时自动忽略重复：

PostgreSQL示例 #

from sqlalchemy.dialects.postgresql import insert

stmt = insert(Category).values([{"text": t} for t in all_cat_texts])
# 当text字段冲突时，什么都不做
stmt = stmt.on_conflict_do_nothing(index_elements=['text'])
db.execute(stmt)
db.commit()

MySQL示例 #

from sqlalchemy.dialects.mysql import insert

stmt = insert(Category).values([{"text": t} for t in all_cat_texts])
# 冲突时不更新任何字段（等同于忽略）
stmt = stmt.on_duplicate_key_update(text=stmt.inserted.text)
db.execute(stmt)
db.commit()

这种方式无需手动查询，直接让数据库处理重复逻辑，效率最高，适合大规模批量操作。

以上方案均可复用至你所有的多对多关联场景，无需逐个手动去重。

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