SQLAlchemy多对多批量插入报错：如何调整事务操作顺序？

问题场景 #

使用SQLAlchemy 2.0.1实现了Player、Match、Elo三张表的多对多关联，模型代码如下：

import os
from datetime import datetime
from uuid import UUID

from pytz import utc
from sqlalchemy import Column, create_engine, DateTime, ForeignKey, Integer
from sqlalchemy.orm import registry, relationship, Mapped, Session
from sqlalchemy import UUID as saUUID


mapper_registry = registry()


@mapper_registry.mapped
class Elo:
    __tablename__ = "elos"

    player: UUID = Column(saUUID, ForeignKey("players.id"), primary_key=True)
    match: UUID = Column(saUUID, ForeignKey("matches.id"), primary_key=True)
    elo: int = Column(Integer)


@mapper_registry.mapped
class Player:
    __tablename__ = "players"

    id: Mapped[UUID] = Column(saUUID, primary_key=True)

    matches: Mapped[list["Match"]] = relationship(secondary="elos", back_populates="players")


@mapper_registry.mapped
class Match:
    __tablename__ = "matches"

    id: Mapped[UUID] = Column(saUUID, primary_key=True)
    date: Mapped[datetime] = Column(DateTime, default=datetime.now(tz=utc))

    players: Mapped[list[Player]] = relationship(secondary="elos", back_populates="matches")

尝试在一个事务中同时插入Player、Match和Elo对象时，触发外键约束报错，错误信息：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table "elos" violates foreign key constraint "elos_player_fkey"

测试代码：

if __name__ == '__main__':
    con_string = "postgresql://user:password@127.0.0.1:5432/db"
    engine = create_engine(con_string, echo=True)
    with Session(engine) as session:
        player = Player(id=UUID(bytes=os.urandom(16)))
        match = Match(id=UUID(bytes=os.urandom(16)))
        elo = Elo(player=player.id, match=match.id, elo=1999)
        session.add_all([player, match, elo])
        session.commit()

推测SQLAlchemy按表名字母顺序执行插入（先处理Elo表），此时Player和Match的记录尚未写入数据库，导致外键约束失败。

解决方案 #

方案1：利用多对多关系自动管理关联表 #

既然已经定义了多对多关系，不需要手动创建Elo对象，直接通过关系属性关联Player和Match，SQLAlchemy会自动处理插入顺序：

if __name__ == '__main__':
    con_string = "postgresql://user:password@127.0.0.1:5432/db"
    engine = create_engine(con_string, echo=True)
    with Session(engine) as session:
        player = Player(id=UUID(bytes=os.urandom(16)))
        match = Match(id=UUID(bytes=os.urandom(16)))
        # 通过关系属性关联，SQLAlchemy自动处理Elo表的插入
        player.matches.append(match)
        session.add_all([player, match])
        session.commit()

如果需要在关联表中存储额外字段（比如elo），则需要使用关联对象模式，调整模型定义。

方案2：改用关联对象模式，让SQLAlchemy识别依赖关系 #

修改模型，让Elo作为关联对象显式关联Player和Match，这样SQLAlchemy能自动识别插入顺序：

# 修改后的模型代码
from sqlalchemy.ext.associationproxy import association_proxy  # 需要导入association_proxy

@mapper_registry.mapped
class Elo:
    __tablename__ = "elos"

    player_id: UUID = Column(saUUID, ForeignKey("players.id"), primary_key=True)
    match_id: UUID = Column(saUUID, ForeignKey("matches.id"), primary_key=True)
    elo: int = Column(Integer)
    
    # 显式定义与Player、Match的关系
    player: Mapped[Player] = relationship(back_populates="elo_entries")
    match: Mapped[Match] = relationship(back_populates="elo_entries")


@mapper_registry.mapped
class Player:
    __tablename__ = "players"

    id: Mapped[UUID] = Column(saUUID, primary_key=True)

    # 关联到Elo对象而非直接关联Match
    elo_entries: Mapped[list["Elo"]] = relationship(back_populates="player")
    # 用association_proxy简化Match的访问（可选）
    matches: Mapped[list["Match"]] = association_proxy("elo_entries", "match")


@mapper_registry.mapped
class Match:
    __tablename__ = "matches"

    id: Mapped[UUID] = Column(saUUID, primary_key=True)
    date: Mapped[datetime] = Column(DateTime, default=datetime.now(tz=utc))

    elo_entries: Mapped[list["Elo"]] = relationship(back_populates="match")
    # 用association_proxy简化Player的访问（可选）
    players: Mapped[list[Player]] = association_proxy("elo_entries", "player")

插入时直接关联对象而非ID，SQLAlchemy会按正确顺序执行插入：

if __name__ == '__main__':
    con_string = "postgresql://user:password@127.0.0.1:5432/db"
    engine = create_engine(con_string, echo=True)
    with Session(engine) as session:
        player = Player(id=UUID(bytes=os.urandom(16)))
        match = Match(id=UUID(bytes=os.urandom(16)))
        elo = Elo(player=player, match=match, elo=1999)
        session.add_all([player, match, elo])
        session.commit()

方案3：手动控制插入顺序（不推荐） #

如果一定要手动插入，可以先刷新会话确保Player和Match写入数据库，再插入Elo：

if __name__ == '__main__':
    con_string = "postgresql://user:password@127.0.0.1:5432/db"
    engine = create_engine(con_string, echo=True)
    with Session(engine) as session:
        player = Player(id=UUID(bytes=os.urandom(16)))
        match = Match(id=UUID(bytes=os.urandom(16)))
        session.add_all([player, match])
        session.flush()  # 刷新会话，确保Player和Match记录写入数据库
        elo = Elo(player=player.id, match=match.id, elo=1999)
        session.add(elo)
        session.commit()

关键原因 #

这个问题的根源是手动创建Elo对象时，只传入了Player和Match的ID而没有建立对象引用，导致SQLAlchemy无法识别它们之间的依赖关系，只能按表名字母顺序执行插入，此时Player和Match的记录尚未写入数据库，触发外键约束错误。

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