# FastAPI 集成

[English](./fastapi_integration.md)

pydantic-resolve 与 FastAPI 自然协作，因为两者都使用 Pydantic 模型。本页面介绍常见的集成模式。

## 目标

你希望 FastAPI 端点返回已解析的数据 —— 关系已加载、派生字段已计算 —— 在单个请求中完成：

```json
[
    {"id": 10, "title": "Design docs", "owner_id": 7, "owner": {"id": 7, "name": "Ada"}},
    {"id": 11, "title": "Refine examples", "owner_id": 8, "owner": {"id": 8, "name": "Bob"}}
]
```

没有 N+1 查询。路由处理器中不需要手动 join 逻辑。

## Step 1：在路由处理器中解析

```python
from fastapi import FastAPI
from pydantic import BaseModel
from pydantic_resolve import Loader, Resolver

app = FastAPI()


class UserView(BaseModel):
    id: int
    name: str


class TaskView(BaseModel):
    id: int
    title: str
    owner_id: int
    owner: Optional[UserView] = None

    def resolve_owner(self, loader=Loader(user_loader)):  # (1)
        return loader.load(self.owner_id)


@app.get("/tasks", response_model=list[TaskView])
async def get_tasks():
    tasks = await fetch_tasks_from_db()
    task_views = [TaskView.model_validate(t) for t in tasks]
    return await Resolver().resolve(task_views)  # (2)
```

1.  `resolve_owner` 声明缺失字段 —— 与快速开始中相同。
2.  `Resolver().resolve()` 遍历模型树并批量加载所有关系。`response_model` 负责序列化。

## Step 2：传递请求上下文

使用 `Resolver(context=...)` 将请求范围的数据传入 `post_*` 方法：

```python
from fastapi import Request


@app.get("/tasks")
async def get_tasks(request: Request):
    user_id = request.state.user_id
    tasks = await fetch_tasks()
    task_views = [TaskView.model_validate(t) for t in tasks]
    return await Resolver(context={
        'user_id': user_id,
        'permissions': ['read', 'write'],
    }).resolve(task_views)


class TaskView(BaseModel):
    owner: Optional[UserView] = None
    can_edit: bool = False

    def resolve_owner(self, loader=Loader(user_loader)):
        return loader.load(self.owner_id)

    def post_can_edit(self, context):  # (1)
        return 'write' in context.get('permissions', [])
```

1.  `context` 是传入 `Resolver()` 的字典。用于权限、区域设置或任何请求范围的数据。

## Step 3：结合 FastAPI 依赖注入与 Loader 参数

```python
from fastapi import Depends, Query


async def get_status_filter(status: str = Query('active')) -> str:
    return status


@app.get("/companies")
async def get_companies(status: str = Depends(get_status_filter)):
    companies = await fetch_companies()
    return await Resolver(
        loader_params={OfficeLoader: {'status': status}}  # (1)
    ).resolve(companies)
```

1.  `loader_params` 将过滤器传给 loader 的批量函数。每个 loader 只接收为它声明的参数。

## 共享 Resolver 配置

当多个端点共享相同配置时，创建一个工厂：

```python
def make_resolver(request: Request) -> Resolver:
    return Resolver(
        context={'user_id': request.state.user_id},
        loader_params={
            OfficeLoader: {'status': 'active'},
        },
    )


@app.get("/tasks")
async def get_tasks(request: Request):
    resolver = make_resolver(request)
    tasks = await fetch_tasks()
    return await resolver.resolve([TaskView.model_validate(t) for t in tasks])


@app.get("/sprints")
async def get_sprints(request: Request):
    resolver = make_resolver(request)
    sprints = await fetch_sprints()
    return await resolver.resolve([SprintView.model_validate(s) for s in sprints])
```

## 错误处理

将 resolver 调用包装在 try/except 中以获得清晰的错误响应：

```python
from pydantic_resolve import LoaderFieldNotProvidedError


@app.get("/tasks")
async def get_tasks():
    try:
        tasks = await fetch_tasks()
        return await Resolver(
            loader_params={OfficeLoader: {'status': 'active'}}
        ).resolve([TaskView.model_validate(t) for t in tasks])
    except LoaderFieldNotProvidedError as e:
        raise HTTPException(status_code=500, detail=str(e))
```

## OpenAPI Schema 生成

FastAPI 自动从 Pydantic 模型生成 OpenAPI schema。以 `None` 和 `Optional` 类型开头的字段会正确显示：

```python
class TaskView(BaseModel):
    id: int
    title: str
    owner_id: int
    owner: Optional[UserView] = None  # 在 OpenAPI 中显示为可空

    def resolve_owner(self, loader=Loader(user_loader)):
        return loader.load(self.owner_id)
```

`owner` 字段在 schema 中显示为 `{"oneOf": [{"type": "null"}, {"$ref": "UserView"}]}`。

如果你希望从输入 schema 中排除已解析的字段，同时保留在输出中，请使用单独的请求/响应模型：

```python
class TaskCreate(BaseModel):
    """输入模型 — 不包含已解析字段"""
    title: str
    owner_id: int


class TaskResponse(BaseModel):
    """输出模型 — 包含已解析字段"""
    id: int
    title: str
    owner_id: int
    owner: Optional[UserView] = None

    def resolve_owner(self, loader=Loader(user_loader)):
        return loader.load(self.owner_id)


@app.post("/tasks", response_model=TaskResponse)
async def create_task(data: TaskCreate):
    task = await create_task_in_db(data)
    task_view = TaskResponse.model_validate(task)
    return await Resolver().resolve(task_view)
```

## 性能

1.  **每个请求一个 `Resolver()`。** resolver 每次都创建新的 DataLoader 实例，因此批次范围正确。

2.  **一次性解析整个列表。** 不要在循环内逐项解析：

    ```python
    # 错误：N 个 resolver 调用
    results = []
    for task in tasks:
        result = await Resolver().resolve(TaskView.model_validate(task))
        results.append(result)

    # 正确：一个 resolver 调用
    task_views = [TaskView.model_validate(t) for t in tasks]
    results = await Resolver().resolve(task_views)
    ```

3.  **使用 `response_model` 进行序列化。** 让 FastAPI 处理 JSON 转换 — 不要手动调用 `model_dump()`。

4.  **调试模式。** 在开发期间启用 `Resolver(debug=True)` 以查看每个节点的计时。

## 下一步

继续阅读 [GraphQL 指南](./graphql_guide.zh.md) 了解如何从 ERD 生成 GraphQL，或 [MCP 服务](./mcp_service.zh.md) 向 AI 代理暴露 API。
