FastAPI+SQLAlchemy多对多关联：重复插入子项触发唯一约束错误

问题描述 #

使用FastAPI和SQLAlchemy实现多对多关系存储时，已存在的子项会被当作新条目插入，触发主键唯一约束错误。首次运行接口可正常创建父项、子项及关联关系；第二次运行时，因尝试插入已存在ID的子项，报错如下：

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: children.id
[SQL: INSERT INTO children (id, name) VALUES (?, ?)]
[parameters: (1, 'Child name')]

复现代码：

from fastapi import Depends, FastAPI
from sqlalchemy import Column, ForeignKey, Integer, create_engine, String, Table
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

children_parents = Table(
    "children_parents",
    Base.metadata,
    Column("child_id", ForeignKey("children.id"), primary_key=True),
    Column("parent_id", ForeignKey("parents.id"), primary_key=True)
)

class Parent(Base):
    __tablename__ = "parents"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

    children = relationship(
        "Child", secondary=children_parents, back_populates="parents")

class Child(Base):
    __tablename__ = "children"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

    parents = relationship(
        "Parent", secondary=children_parents, back_populates="children")

app = FastAPI()

Base.metadata.create_all(bind=engine)

@app.get("/")
async def root(db: SessionLocal = Depends(get_db)):
    parent = Parent(name="Parent name", children=[
        Child(id=1, name="Child name")])
    db.add(parent)
    db.commit()
    db.refresh(parent)
    return parent

解决方案 #

方案1：先查询复用已存在的子项 #

在创建父项前，先查询数据库中是否存在目标子项，存在则直接使用该实例，不存在再创建新实例：

@app.get("/")
async def root(db: SessionLocal = Depends(get_db)):
    # 查询是否存在id为1的子项
    child = db.query(Child).filter(Child.id == 1).first()
    if not child:
        child = Child(id=1, name="Child name")
        db.add(child)
        db.commit()
        db.refresh(child)
    
    # 创建父项并关联子项
    parent = Parent(name="Parent name", children=[child])
    db.add(parent)
    db.commit()
    db.refresh(parent)
    return parent

方案2：使用session.merge()自动处理实例 #

merge()方法会根据主键判断实例是否存在：存在则更新属性并关联，不存在则插入新实例：

@app.get("/")
async def root(db: SessionLocal = Depends(get_db)):
    # 用merge处理子项，自动识别是否已存在
    child = db.merge(Child(id=1, name="Child name"))
    
    parent = Parent(name="Parent name", children=[child])
    db.add(parent)
    db.commit()
    db.refresh(parent)
    return parent

关键原因说明 #

直接创建Child(id=1, name="Child name")实例时，SQLAlchemy的session会将其标记为新对象，即使id已存在。必须通过查询让session加载已存在的实例，或使用merge()让session自动识别实例状态，避免重复插入。

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