SQLAlchemy使用bulk_save_objects时Row记录因关联标签重复的解决方法

提交关联了带标签行的Header对象时，每个Row会按其关联标签的数量重复插入。已知SQLAlchemy将此视为关联表RowTag的新记录，进而导致Row记录重复。需要避免Row记录重复，确保SQLAlchemy识别现有标签且不创建重复。

以下是场景中的Schema及示例代码：

数据库Schema #

from sqlalchemy import Column, Integer, ForeignKey
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import relationship

class MySQLAlchemy(SQLAlchemy):
    """A custom SQLAlchemy class to handle multiple databases sessions."""

db = MySQLAlchemy()
Model = db.Model

class Header(Model):
    __tablename__ = 'header'
    id = Column(Integer, primary_key=True)

class Row(Model):
    __tablename__ = 'row'
    id = Column(Integer, primary_key=True)
    header_id = Column(Integer)
    tags = relationship("Tag", secondary="row_tag", back_populates="rows")

class Tag(Model):
    __tablename__ = 'tag'
    id = Column(Integer, primary_key=True)
    rows = relationship("Row", secondary="row_tag", back_populates="tags")

class RowTag(Model):
    __tablename__ = 'row_tag'
    row_id = Column(Integer, ForeignKey('row.id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tag.id'), primary_key=True)

示例业务代码 #

def insert_rows(header_id, tags):
    """
        Simplified version of inserting rows function,
        more rows attributes come with the request, instead of doing a dummy loop.
    """
    ready_rows = []
    for i in range(3):
        row1 = Row({"id": i, "header_id": header_id})
        row1.tags.extend(tags)
        ready_rows.append(row1)
        # 若取消以下两行注释，Row可正常插入且无重复，每行关联2个标签
        # session.add(row1)
        # session.flush()
    return ready_rows

def new_page():
    # 从数据库中查询标签
    tags = [tag1, tag2]

    # 添加Header
    header = Header({"id": 1})
    session.add(header)
    session.flush()

    # 添加Row
    ready_rows = insert_rows(header.id, tags)

    # 此处使用bulk_save_objects添加Row，每个Row会被插入两次（即标签数量次）
    session.bulk_save_objects(ready_rows)
    session.flush()   

    # 最终提交
    session.commit()
    return None, 200

# 执行
new_page()

1. 拆分操作：先插Row，再关联Tag #

先批量插入所有Row，让它们进入持久化状态，再为每个Row关联Tag。这样关联操作只会生成RowTag的关联记录，不会重复插入Row。

修改后的代码：

def insert_rows(header_id):
    ready_rows = []
    for i in range(3):
        row1 = Row({"id": i, "header_id": header_id})
        ready_rows.append(row1)
    return ready_rows

def new_page():
    # 查询现有标签
    tags = [tag1, tag2]

    # 插入Header
    header = Header({"id": 1})
    session.add(header)
    session.flush()

    # 第一步：批量插入Row
    ready_rows = insert_rows(header.id)
    session.bulk_save_objects(ready_rows)
    session.flush()  # 刷新会话，让Row变为持久化状态

    # 第二步：为Row关联标签
    for row in ready_rows:
        row.tags.extend(tags)

    session.commit()
    return None, 200

2. 用session.add_all()替代bulk_save_objects #

add_all()会让会话跟踪对象状态，自动处理关联关系，不会因为添加多个标签而重复插入Row。虽然性能略低于批量操作，但适配场景更广泛。

修改后的代码：

def insert_rows(header_id, tags):
    ready_rows = []
    for i in range(3):
        row1 = Row({"id": i, "header_id": header_id})
        row1.tags.extend(tags)
        ready_rows.append(row1)
    return ready_rows

def new_page():
    tags = [tag1, tag2]

    header = Header({"id": 1})
    session.add(header)
    session.flush()

    ready_rows = insert_rows(header.id, tags)
    session.add_all(ready_rows)  # 替换bulk_save_objects
    session.commit()

    return None, 200

3. 手动将Row加入会话后再关联标签 #

如果必须使用bulk_save_objects，可以在添加标签前先把Row加入会话并刷新，让SQLAlchemy识别它们是已存在的对象：

修改insert_rows函数：

def insert_rows(header_id, tags, session):
    ready_rows = []
    for i in range(3):
        row1 = Row({"id": i, "header_id": header_id})
        session.add(row1)
        session.flush()  # 刷新后Row进入持久化状态
        row1.tags.extend(tags)
        ready_rows.append(row1)
    return ready_rows

def new_page():
    tags = [tag1, tag2]

    header = Header({"id": 1})
    session.add(header)
    session.flush()

    ready_rows = insert_rows(header.id, tags, session)
    # 无需再调用bulk_save_objects，Row已经被会话跟踪
    session.commit()

    return None, 200

bulk_save_objects是SQLAlchemy的批量操作API，它绕过了会话的对象状态管理，直接生成SQL执行，无法处理关联关系带来的对象状态变化。当Row处于临时状态（未加入会话或未持久化）时，每添加一个标签，SQLAlchemy可能会将其视为新对象重复插入；只有当Row进入持久化状态（已插入数据库并被会话跟踪）后，关联标签才只会生成RowTag的关联记录。

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