如何在SQLAlchemy中插入多对多关系数据？解决唯一约束冲突

场景 #

尝试将以下JSON数据插入PostgreSQL数据库的SQLAlchemy多对多关系中：

[{
    "name": "CORVETTE",
    "officers": [{"first_name": "Alan1", "rank": "Commander"}, 
             {"first_name": "Alan2", "rank": "Soldier"}]
  },
  {
    "name": "CRUISER",
    "officers": [{"first_name": "Alan1", "rank": "Commander"}, 
             {"first_name": "Alan3", "rank": "Soldier"}]
  }
]

对应的SQLAlchemy模型定义：

class Officer_orm(Base):
    __tablename__ = "officer_orm"
    first_name = Column(String(255), primary_key=True)
    rank = Column(String(255), primary_key=True)

    spaceships_orm = relationship('Spaceship_orm', 
                      secondary='spaceships_officers', 
                      lazy='subquery', 
                      back_populates='officers_orm')


class Spaceship_orm(Base):
    __tablename__ = "spaceship_orm"
    name = Column(String(255), primary_key=True)

    officers_orm = relationship('Officer_orm', 
                    secondary='spaceships_officers', 
                    lazy='subquery', 
                    back_populates='spaceships_orm')            


class SpaceshipsOfficers(Base):
    __tablename__ = "spaceships_officers"

    id = Column(Integer, primary_key=True)
    o_first_name = Column(String(255))
    o_rank = Column(String(255))
    s_name = Column(String(255), ForeignKey('spaceship_orm.name'))
    __table_args__ = ((ForeignKeyConstraint(["o_first_name", "o_rank"], 
                          ["officer_orm.first_name","officer_orm.rank"])),) 

插入代码：

# .... fill spaceships from json
for s in json_spaceships:
    # .... fill spaceships.officers from json
    with Session(engine) as session:
        spaceship_orm = Spaceship_orm(name=s.name)
        for o in officers:
            officer_orm = Officer_orm(first_name=o.first_name, rank=o.rank)
            spaceship_orm.officers_orm.append(officer_orm)

        session.add_all([spaceship_orm])
        session.commit()

错误信息 #

插入第二个飞船CRUISER时触发错误：

ERROR:root:(psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "officer_orm_pkey" DETAIL:  Key (first_name,rank)=(Alan1, Commander) already exists.

问题原因 #

Alan1（军衔Commander）在插入第一个飞船CORVETTE时已存入officer_orm表，由于该表主键是first_name+rank的组合键，再次创建同名同军衔的Officer_orm实例并插入时，违反了主键唯一约束。

解决方案 #

核心思路：先查询数据库中是否已存在该军官，存在则直接关联，不存在再创建新实例。

方法1：手动实现get_or_create逻辑 #

修改插入代码，对每个军官先尝试从数据库查询，不存在则创建：

# 统一使用一个会话处理所有插入，避免多次会话导致的实例分离
with Session(engine) as session:
    for s in json_spaceships:
        spaceship_orm = Spaceship_orm(name=s["name"])
        session.add(spaceship_orm)
        
        for o in s["officers"]:
            # 按主键查询是否存在该军官
            existing_officer = session.query(Officer_orm).filter_by(
                first_name=o["first_name"], rank=o["rank"]
            ).first()
            
            if existing_officer:
                # 关联已有军官
                spaceship_orm.officers_orm.append(existing_officer)
            else:
                # 创建新军官并关联
                new_officer = Officer_orm(first_name=o["first_name"], rank=o["rank"])
                session.add(new_officer)
                spaceship_orm.officers_orm.append(new_officer)
    
    session.commit()

方法2：使用SQLAlchemy的merge方法 #

merge方法会自动判断实例是否存在于数据库，存在则返回现有实例，不存在则创建新实例：

with Session(engine) as session:
    for s in json_spaceships:
        spaceship_orm = Spaceship_orm(name=s["name"])
        session.add(spaceship_orm)
        
        for o in s["officers"]:
            officer_orm = Officer_orm(first_name=o["first_name"], rank=o["rank"])
            # merge自动处理存在/不存在的情况
            merged_officer = session.merge(officer_orm)
            spaceship_orm.officers_orm.append(merged_officer)
    
    session.commit()

额外优化：调整会话范围 #

原代码每次循环创建新会话，导致不同会话中的实例无法共享，统一使用一个会话处理所有插入操作，能减少数据库交互，也避免跨会话的实例状态问题。

目标关联数据 #

执行后spaceships_officers表应生成如下数据：

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