FastAPI 应用

演示如何用 FastAPI + Pydantic 构建一个极简的 Agent HTTP 服务， 包含路径参数、查询参数、请求体校验、错误处理与异步路由。 对应教程：第2卷《Python 进阶》第4章 FastAPI 与 Pydantic。 运行方式:: uvicorn examples.volume-02-python.fastapi_app:app --reload

文件路径: examples/volume-02-python\fastapi_app.py

---

源代码

"""FastAPI 服务示例。

演示如何用 FastAPI + Pydantic 构建一个极简的 Agent HTTP 服务，
包含路径参数、查询参数、请求体校验、错误处理与异步路由。

对应教程：第2卷《Python 进阶》第4章 FastAPI 与 Pydantic。

运行方式::

    uvicorn examples.volume-02-python.fastapi_app:app --reload
"""

from __future__ import annotations

from typing import Annotated

from fastapi import FastAPI, HTTPException, Query, status
from pydantic_models import AgentConfig, ChatLog, Message, RoleEnum

app = FastAPI(title="示例 Agent 服务", version="0.1.0")

# 内存存储，便于演示（生产环境应使用数据库）
_chat_logs: dict[str, ChatLog] = {}


@app.get("/health")
def health() -> dict[str, str]:
    """健康检查接口。

    Returns:
        服务状态字典。
    """
    return {"status": "ok"}


@app.post("/sessions/{session_id}", status_code=status.HTTP_201_CREATED)
def create_session(session_id: str, system_prompt: str = "你是一个友好的助手。") -> ChatLog:
    """创建一个会话。

    Args:
        session_id: 会话 ID。
        system_prompt: 系统提示词。

    Returns:
        新建的 ``ChatLog``。

    Raises:
        HTTPException: 会话已存在时返回 409。
    """
    if session_id in _chat_logs:
        raise HTTPException(status_code=409, detail=f"会话 {session_id} 已存在")
    log = ChatLog(
        session_id=session_id,
        messages=[Message(role=RoleEnum.SYSTEM, content=system_prompt)],
    )
    _chat_logs[session_id] = log
    return log


@app.get("/sessions/{session_id}")
def get_session(session_id: str) -> ChatLog:
    """获取会话内容。

    Args:
        session_id: 会话 ID。

    Returns:
        对应的 ``ChatLog``。

    Raises:
        HTTPException: 会话不存在时返回 404。
    """
    log = _chat_logs.get(session_id)
    if log is None:
        raise HTTPException(status_code=404, detail=f"会话 {session_id} 不存在")
    return log


@app.post("/sessions/{session_id}/messages")
def add_message(session_id: str, message: Message) -> ChatLog:
    """向会话追加一条消息。

    Args:
        session_id: 会话 ID。
        message: 待追加的消息。

    Returns:
        更新后的 ``ChatLog``。
    """
    log = _chat_logs.get(session_id)
    if log is None:
        raise HTTPException(status_code=404, detail=f"会话 {session_id} 不存在")
    log.messages.append(message)
    return log


@app.get("/sessions")
def list_sessions(
    limit: Annotated[int, Query(ge=1, le=100, description="返回数量上限")] = 10,
) -> list[ChatLog]:
    """列出最近创建的若干会话。

    Args:
        limit: 返回数量上限。

    Returns:
        ``ChatLog`` 列表。
    """
    return list(_chat_logs.values())[:limit]


@app.delete("/sessions/{session_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_session(session_id: str) -> None:
    """删除会话。

    Args:
        session_id: 会话 ID。

    Raises:
        HTTPException: 会话不存在时返回 404。
    """
    if session_id not in _chat_logs:
        raise HTTPException(status_code=404, detail=f"会话 {session_id} 不存在")
    del _chat_logs[session_id]


@app.post("/validate/config")
def validate_config(config: AgentConfig) -> dict[str, object]:
    """校验 Agent 配置并返回摘要。

    Args:
        config: 待校验的配置。

    Returns:
        包含 ``valid`` 与 ``config`` 的字典。
    """
    return {"valid": True, "config": config.model_dump()}


def reset_store() -> None:
    """清空内存存储（测试辅助）。"""
    _chat_logs.clear()


def main() -> None:
    """演示用 TestClient 本地调用服务。"""
    import uvicorn  # 延迟导入，仅在直接运行时需要

    uvicorn.run(app, host="127.0.0.1", port=8000)


if __name__ == "__main__":
    main()