FastAPI+SQLAlchemy如何返回带嵌套子分类的ORM模型结果

当前返回结果不符合嵌套结构的核心原因：

• 查询仅选取部分字段，未加载child关联关系，且返回原始行数据而非模型实例
• SQLAlchemy自关联关系的remote_side配置错误，无法正确关联子分类
• Pydantic模型的child字段类型定义错误，不支持递归嵌套结构
• 查询未过滤根分类，导致子分类被当作独立条目返回

1. 修复SQLAlchemy自关联关系配置 #

修正remote_side指向父分类的id字段，确保自关联逻辑正确：

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from database._types import created_at, uuid, uuidpk
from database.engine import Base


class CategoriesModel(Base):
    __tablename__ = "categories"

    id: Mapped[uuidpk]
    parent_id: Mapped[uuid | None] = mapped_column(ForeignKey("categories.id"))
    name: Mapped[str]
    created_at: Mapped[created_at]

    child: Mapped[list["CategoriesModel"]] = relationship(
        "CategoriesModel",
        remote_side=[id],  # 修正为指向父分类的id字段
        uselist=True,
        lazy="selectin"  # 默认懒加载方式，也可在查询时指定
    )

2. 修改仓库查询逻辑，加载嵌套关系并过滤根分类 #

查询完整模型实例，预加载子分类，仅返回根分类（parent_id为None）：

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession

class SQLAlchemyRepository[T]:
    def __init__(self, session: AsyncSession) -> None:
        self._model = get_args(self.__orig_bases__[0])[0]
        self._session = session


class CategoriesRepository(SQLAlchemyRepository[CategoriesModel]):
    async def get_many(self) -> list[CategoriesModel]:
        stmt = (
            select(self._model)
            .where(self._model.parent_id.is_(None))  # 仅查询根分类
            .order_by(self._model.name)
            .options(selectinload(self._model.child))  # 预加载子分类，避免N+1查询
        )
        res = await self._session.execute(stmt)
        return res.scalars().all()  # 返回模型实例列表，而非Row对象

3. 修正Pydantic模型的递归嵌套定义 #

让child字段递归引用自身，支持嵌套结构序列化：

from datetime import datetime
from pydantic import BaseModel
from _types import uuid


class SQLAlchemyMappedModel(BaseModel):
    class Config:
        from_attributes = True


class CategoryOut(SQLAlchemyMappedModel):
    id: uuid
    name: str
    created_at: datetime
    child: list["CategoryOut"] | None = None  # 递归引用自身，支持多层嵌套

# 解决递归引用的类型解析问题
CategoryOut.model_rebuild()

4. 调整FastAPI接口（优化类型匹配） #

显式指定响应模型，移除不必要的提交操作：

from fastapi import APIRouter, Depends, Annotated
from your_uow_module import IUnitOfWork, get_uow
from your_pydantic_module import CategoryOut

router = APIRouter()

@router.get("/", summary="Categories", response_model=list[CategoryOut])
async def get_categories(
    uow: Annotated[IUnitOfWork, Depends(get_uow)],
):
    async with uow:
        res = await uow.categories.get_many()
    return res

修改完成后，接口将返回符合预期的嵌套JSON结构：根分类包含对应的子分类，无子分类的条目child字段为None。

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