使用PostgreSQL on_conflict_do_update时SQLAlchemy多对多关系无法提交

你用SQLAlchemy Core的INSERT ... ON CONFLICT DO UPDATE创建对象后，ORM会话根本不知道这个对象的存在——因为Core操作绕过了ORM的会话缓存和跟踪机制。所以你调用relationship.append()时，ORM无法识别这个关联变更，自然不会向关联表插入数据。而常规ORM对象创建+append的方式重复执行会触发唯一约束，是因为没处理冲突逻辑。

方案1：让ORM会话接管Core创建的对象 #

通过INSERT语句的returning()拿到创建/更新后的对象数据，再用session.merge()把对象纳入会话管理。这样后续的append()操作就能被ORM跟踪，提交时自动生成关联表的插入语句。同时给关联表加唯一约束，配合冲突处理逻辑避免重复提交报错。

代码示例

先定义模型（对应你的B、C和关联表BC）：

from sqlalchemy import Column, Integer, String, Table, ForeignKey, UniqueConstraint
from sqlalchemy.orm import declarative_base, relationship, Session
from sqlalchemy.dialects.postgresql import insert

Base = declarative_base()

# 多对多关联表
bc_association = Table(
    'bc', Base.metadata,
    Column('b_id', Integer, ForeignKey('b.id'), primary_key=True),
    Column('c_id', Integer, ForeignKey('c.id'), primary_key=True),
    UniqueConstraint('b_id', 'c_id', name='bc_unique')
)

class B(Base):
    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True, nullable=False)
    cs = relationship('C', secondary=bc_association, back_populates='bs')

class C(Base):
    __tablename__ = 'c'
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True, nullable=False)
    bs = relationship('B', secondary=bc_association, back_populates='cs')

然后执行操作：

with Session(engine) as session:
    # 1. 插入/更新B对象，返回实例
    insert_b = insert(B).values(name='b1').on_conflict_do_update(
        index_elements=[B.name],
        set_=dict(name='b1')
    ).returning(B)
    b_obj = session.scalar(insert_b)
    # 关键：把Core创建的对象合并到会话，让ORM跟踪它
    b_obj = session.merge(b_obj)

    # 2. 同理处理C对象
    insert_c = insert(C).values(name='c1').on_conflict_do_update(
        index_elements=[C.name],
        set_=dict(name='c1')
    ).returning(C)
    c_obj = session.scalar(insert_c)
    c_obj = session.merge(c_obj)

    # 3. 添加关联，此时会话能识别变更
    # 可选：先检查关联是否已存在，避免重复append
    if b_obj not in c_obj.bs:
        c_obj.bs.append(b_obj)

    session.commit()

方案2：全Core操作保证极致原子性 #

如果要彻底避免ORM会话的潜在问题，直接用Core语句同时处理主对象和关联表的操作，全程原子性，且重复执行不会报错：

with Session(engine) as session:
    # 插入/更新B，获取ID
    insert_b = insert(B).values(name='b1').on_conflict_do_update(
        index_elements=[B.name],
        set_=dict(name='b1')
    ).returning(B.id)
    b_id = session.scalar(insert_b)

    # 插入/更新C，获取ID
    insert_c = insert(C).values(name='c1').on_conflict_do_update(
        index_elements=[C.name],
        set_=dict(name='c1')
    ).returning(C.id)
    c_id = session.scalar(insert_c)

    # 插入关联，冲突则忽略
    insert_bc = insert(bc_association).values(b_id=b_id, c_id=c_id).on_conflict_do_nothing(
        index_elements=['b_id', 'c_id']
    )
    session.execute(insert_bc)

    session.commit()

• 方案1的session.merge()是核心：它会把数据库返回的对象实例同步到会话缓存，让ORM能够跟踪后续的关联变更。
• 方案2完全绕开ORM的关系跟踪，直接操作数据库，原子性最强，适合高性能或批量场景。
• 不管哪种方案，关联表必须有(b_id, c_id)的唯一约束，否则重复插入会导致重复数据。

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