如何在SQLModel中关联已有父表条目与子表条目避免重复

当前代码的核心问题是：

• 创建子条目时，直接实例化新的Parent对象，而非查询数据库中已存在的同名父条目，导致重复创建父条目。
• 创建父条目时，处理子条目逻辑同理，直接新建Child对象，未复用已有子条目。
• 缺少对已有条目关联关系的更新逻辑。

我们需要调整接口逻辑，实现先查询已有条目，不存在再创建，存在则直接关联的核心逻辑，同时补充关联更新的处理：

1. 优化辅助函数 #

新增一个根据名称获取已有条目的函数，代替单纯的存在检查，减少重复数据库查询：

def get_existing_entry(table: Union[Parent, Child], name: str) -> Optional[Union[Parent, Child]]:
    """根据名称查询数据库中已存在的条目，不存在则返回None"""
    with Session(engine) as session:
        try:
            return session.exec(select(table).where(table.name == name)).one()
        except Exception:
            return None

2. 修改/parent/接口：关联已有子条目 #

修改父条目创建/更新逻辑，传入子条目名称时，优先复用数据库中已有的子条目，不存在再创建新条目，同时支持更新已有父条目的子关联：

@app.post("/parent/")
def create_parent(
        name: str,
        children: List[str] = [],
) -> Parent:
    with Session(engine) as session:
        # 获取已有父条目，不存在则创建
        parent = get_existing_entry(Parent, name)
        if not parent:
            parent = Parent(name=name)
            session.add(parent)
        
        # 处理子条目关联
        for child_name in children:
            child = get_existing_entry(Child, child_name)
            if not child:
                child = Child(name=child_name)
                session.add(child)
            # 避免重复添加关联
            if child not in parent.children:
                parent.children.append(child)
        
        session.commit()
        session.refresh(parent)
        return parent

3. 修改/child/接口：关联已有父条目 #

修改子条目创建/更新逻辑，传入父条目名称时，优先查询复用已有父条目，不存在可选择创建新父条目或返回错误（以下代码保留创建新父条目的逻辑）：

@app.post("/child/")
def create_child(
    name: str,
    parent: Optional[str] = None,
) -> Child:
    with Session(engine) as session:
        # 获取已有子条目，不存在则创建
        child = get_existing_entry(Child, name)
        if not child:
            child = Child(name=name)
            session.add(child)
        
        # 处理父条目关联
        if parent:
            existing_parent = get_existing_entry(Parent, parent)
            if existing_parent:
                child.parent = existing_parent
            else:
                # 父条目不存在时创建新的，可根据需求改为返回404错误
                new_parent = Parent(name=parent)
                session.add(new_parent)
                child.parent = new_parent
        
        session.commit()
        session.refresh(child)
        return child

4. 可选：新增关联更新接口 #

如果需要支持后续单独修改已有条目的关联关系（比如给已存在的子条目绑定父条目），可以新增以下接口：

@app.put("/child/{child_name}/assign-parent/")
def assign_parent_to_child(child_name: str, parent_name: str) -> Child:
    with Session(engine) as session:
        child = get_existing_entry(Child, child_name)
        if not child:
            raise HTTPException(status_code=404, detail="Child not found")
        
        parent = get_existing_entry(Parent, parent_name)
        if not parent:
            raise HTTPException(status_code=404, detail="Parent not found")
        
        child.parent = parent
        session.commit()
        session.refresh(child)
        return child

注意：需要导入HTTPException：from fastapi import FastAPI, HTTPException

整合所有修改后的完整代码如下：

from sqlmodel import SQLModel, Session, Field, Relationship, create_engine, select
from fastapi import FastAPI, HTTPException
from typing import Optional, List, Union

# Define tables and relationships
class Parent(SQLModel, table=True):  # type: ignore
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field()

    children: List["Child"] = Relationship(
        back_populates="parent",
        sa_relationship_kwargs={"cascade": "delete"},
    )

class Child(SQLModel, table=True):  # type: ignore
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field()

    parent: Optional["Parent"] = Relationship(
        back_populates="children",
    )
    parent_id: Optional[int] = Field(
        foreign_key="parent.id",
        default=None,
    )

# Set up database URL
db_name = "database.db"
db_url = f"sqlite:///{db_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(db_url, echo=True, connect_args=connect_args)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

app = FastAPI()

# Initialise database on FastAPI startup
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# Helper functions
def get_existing_entry(table: Union[Parent, Child], name: str) -> Optional[Union[Parent, Child]]:
    """根据名称查询数据库中已存在的条目，不存在则返回None"""
    with Session(engine) as session:
        try:
            return session.exec(select(table).where(table.name == name)).one()
        except Exception:
            return None

@app.post("/parent/")
def create_parent(
        name: str,
        children: List[str] = [],
) -> Parent:
    with Session(engine) as session:
        # 获取已有父条目，不存在则创建
        parent = get_existing_entry(Parent, name)
        if not parent:
            parent = Parent(name=name)
            session.add(parent)
        
        # 处理子条目关联
        for child_name in children:
            child = get_existing_entry(Child, child_name)
            if not child:
                child = Child(name=child_name)
                session.add(child)
            # 避免重复添加关联
            if child not in parent.children:
                parent.children.append(child)
        
        session.commit()
        session.refresh(parent)
        return parent

@app.post("/child/")
def create_child(
    name: str,
    parent: Optional[str] = None,
) -> Child:
    with Session(engine) as session:
        # 获取已有子条目，不存在则创建
        child = get_existing_entry(Child, name)
        if not child:
            child = Child(name=name)
            session.add(child)
        
        # 处理父条目关联
        if parent:
            existing_parent = get_existing_entry(Parent, parent)
            if existing_parent:
                child.parent = existing_parent
            else:
                # 父条目不存在时创建新的，可根据需求改为返回404错误
                new_parent = Parent(name=parent)
                session.add(new_parent)
                child.parent = new_parent
        
        session.commit()
        session.refresh(child)
        return child

@app.put("/child/{child_name}/assign-parent/")
def assign_parent_to_child(child_name: str, parent_name: str) -> Child:
    with Session(engine) as session:
        child = get_existing_entry(Child, child_name)
        if not child:
            raise HTTPException(status_code=404, detail="Child not found")
        
        parent = get_existing_entry(Parent, parent_name)
        if not parent:
            raise HTTPException(status_code=404, detail="Parent not found")
        
        child.parent = parent
        session.commit()
        session.refresh(child)
        return child

• 复用已有条目：所有关联操作前先查询数据库，存在则直接关联，不存在再创建新条目，避免重复数据。
• 减少重复查询：用get_existing_entry统一处理条目查询，降低数据库交互次数。
• 支持关联更新：接口不仅支持创建时绑定关联，还支持后续修改已有条目的关联关系。
• 逻辑简化：去掉原代码中冗余的存在检查和重复查询，逻辑更清晰。

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