基于FastAPI实现递归家族树的问题求助

一、修复SQLAlchemy模型的自关联关系 #

你的children字段无法同时对应father和mother两个关系，SQLAlchemy不允许一个关系同时反向关联多个字段，需要拆分或重新定义关系：

from sqlalchemy import Column, Integer, String, ForeignKey, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class DBPerson(Base):
    __tablename__ = 'person'

    id = Column(Integer, primary_key=True)
    name = Column(String, index=True)
    # 其他字段（如性别、出生日期等）...

    father_id = Column(Integer, ForeignKey("person.id"), nullable=True)
    # 指定foreign_keys区分自关联字段，反向关联作为父亲的孩子
    father = relationship("DBPerson", foreign_keys=[father_id], back_populates="father_children")
    
    mother_id = Column(Integer, ForeignKey("person.id"), nullable=True)
    # 指定foreign_keys区分自关联字段，反向关联作为母亲的孩子
    mother = relationship("DBPerson", foreign_keys=[mother_id], back_populates="mother_children")

    # 分别存储作为父亲/母亲的孩子（可根据需求选择是否保留）
    father_children = relationship("DBPerson", foreign_keys=[father_id], back_populates="father")
    mother_children = relationship("DBPerson", foreign_keys=[mother_id], back_populates="mother")

    # 可选：统一的children字段（仅用于查询，不支持写入）
    children = relationship(
        "DBPerson",
        primaryjoin="or_(DBPerson.father_id == foreign(DBPerson.id), DBPerson.mother_id == foreign(DBPerson.id))",
        viewonly=True
    )

二、解决Pydantic模型的递归引用问题 #

直接在类定义中引用自身会导致未定义错误，可通过字符串类型注解解决，这是Pydantic v1/v2通用的方案：

from pydantic import BaseModel

class PersonBase(BaseModel):
    id: int
    name: str
    father: "PersonBase | None" = None
    mother: "PersonBase | None" = None

    class Config:
        orm_mode = True  # Pydantic v1；v2改用from_attributes=True

# Pydantic v1需要手动更新引用（v2可省略，但加上更保险）
PersonBase.update_forward_refs()

三、限制递归世代数 #

直接递归序列化可能导致无限循环或返回过多数据，可通过两种方式控制深度：

方式1：自定义序列化函数（灵活直观） #

编写递归函数，传入深度参数，达到指定深度后停止返回父母信息：

def serialize_person(person: DBPerson, max_depth: int = 3) -> dict:
    """
    将DBPerson对象序列化为家族树结构，限制最大递归深度
    :param max_depth: 最大返回世代数（0表示仅返回当前节点，1表示包含父母，以此类推）
    """
    base_data = {"id": person.id, "name": person.name}
    
    if max_depth <= 0:
        base_data["father"] = None
        base_data["mother"] = None
        return base_data
    
    # 递归序列化父亲（如果存在）
    base_data["father"] = serialize_person(person.father, max_depth - 1) if person.father else None
    # 递归序列化母亲（如果存在）
    base_data["mother"] = serialize_person(person.mother, max_depth - 1) if person.mother else None
    
    return base_data

在FastAPI路由中使用：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from your_models import DBPerson, get_db  # 替换为你的数据库依赖

app = FastAPI()

@app.get("/person/{person_id}")
def get_family_tree(person_id: int, depth: int = 2, db: Session = Depends(get_db)):
    person = db.query(DBPerson).filter(DBPerson.id == person_id).first()
    if not person:
        return {"error": "Person not found"}
    return serialize_person(person, max_depth=depth)

方式2：SQL递归CTE查询（性能更优） #

通过SQL递归公共表表达式（CTE）预查询指定深度内的所有家族成员，再构建树结构，避免N+1查询问题：

def get_family_tree_with_cte(db: Session, person_id: int, max_depth: int = 3):
    # 初始化递归CTE，从指定节点开始，深度为0
    cte = db.query(
        DBPerson.id,
        DBPerson.name,
        DBPerson.father_id,
        DBPerson.mother_id,
        text("0 as depth")
    ).filter(DBPerson.id == person_id).cte(recursive=True)

    # 递归查询父母节点，深度递增，直到达到max_depth
    cte_recursive = db.query(
        DBPerson.id,
        DBPerson.name,
        DBPerson.father_id,
        DBPerson.mother_id,
        text(f"{cte.c.depth} + 1 as depth")
    ).join(cte, DBPerson.id == cte.c.father_id).union_all(
        db.query(
            DBPerson.id,
            DBPerson.name,
            DBPerson.father_id,
            DBPerson.mother_id,
            text(f"{cte.c.depth} + 1 as depth")
        ).join(cte, DBPerson.id == cte.c.mother_id)
    ).filter(cte.c.depth < max_depth)

    # 将所有查询结果存入字典，便于快速查找
    node_map = {}
    for row in db.query(cte.union_all(cte_recursive)).all():
        node_map[row.id] = {
            "id": row.id,
            "name": row.name,
            "father_id": row.father_id,
            "mother_id": row.mother_id,
            "depth": row.depth
        }

    # 递归构建家族树
    def build_tree(node_id):
        node = node_map.get(node_id)
        if not node:
            return None
        # 达到最大深度时，不再递归返回父母
        if node["depth"] >= max_depth:
            return {"id": node["id"], "name": node["name"], "father": None, "mother": None}
        return {
            "id": node["id"],
            "name": node["name"],
            "father": build_tree(node["father_id"]),
            "mother": build_tree(node["mother_id"])
        }

    return build_tree(person_id)

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