第 2 课：FastAPI / Service 分层 —— 工程项目的脊椎骨

本节目标：理解为什么要分层、每层的职责边界在哪、一个HTTP请求从头到尾怎么穿过各层、以及分层之后换数据库为什么只改一个目录。

如果你只能从这门课带走一样东西，就是这个。分层不会让你写新功能更快，但会让你的项目 3 个月后还能改得动。

1. 为什么必须分层

代码不分层 ≈ 把厨房、卧室、厕所、客厅全塞进一个房间。

软件里有 4 种"功能区"，每种变化频率完全不同：

关注点	谁变？	频率
网络协议（HTTP、参数校验、状态码）	前端要新字段、URL 改路径	经常变
业务规则（怎么算账、怎么决策）	产品经理改逻辑	经常变
数据存储（SQLite/Postgres/SQL）	选型变更、迁移	很少变
外部依赖（OpenAI、Chroma、邮件）	换供应商、改 SDK	偶尔变

不分层的代价：HTTP 解析 + 业务规则 + SQL 全写在一个函数里。前端改个字段名 → 你得在 SQL 里跟着改。换数据库 → 你得改路由文件。

核心原则：每层只对一种变化负责，改 A 不影响 B。

2. 实际项目目录结构

Code

app/
├── api/         ← 路由层       管 HTTP 协议
├── services/    ← 服务层       管业务规则
├── db/          ← 数据层       管 SQL
├── models/      ← 数据模型     管"数据形状"
├── core/        ← 公共基础     配置、异常、日志
├── tools/       ← 工具注册     第1课讲过
├── prompts/     ← Prompt 模板  LLM 的"剧本"
├── graph/       ← LangGraph    工作流
├── agents/      ← Agent 编排
├── mcp/         ← MCP 适配
└── deps.py      ← FastAPI 依赖注入

经典 3 层是：api/ → services/ → db/。其他都是配套设施。

3. 每一层的"职责圣旨"

Layer 1 — `api/` 路由层

只做 3 件事：解析 HTTP 请求 → 调 service → 包成响应返回。

python

@router.post("/chat")
async def chat(req: ChatRequest) -> ChatResponse:
    if not req.question.strip():
        raise ValidationError("question 不能只是空白字符")

    logger.info("[chat] request received, q_len=%d", len(req.question))
    result = await llm_service.chat(
        req.question,
        system_prompt=req.system_prompt,
        temperature=req.temperature,
    )
    return ChatResponse(**result)

13 行干了 4 件事：校验输入 → 打日志 → 调 service → 包响应。

注意：铁律：路由层不调 OpenAI、不写 SQL、不写复杂业务逻辑。路由函数尽量 ≤ 20 行。

Layer 2 — `services/` 服务层

业务大脑。所有"该怎么做"的决策都在这层。

python

class LLMService:
    """OpenAI 兼容 API 薄封装.

    可以被 HTTP API、CLI、定时任务、MCP server 调用，完全无感。
    """

    def __init__(self):
        self._client = AsyncOpenAI(
            api_key=settings.openai_api_key,
            base_url=settings.openai_base_url,
            timeout=httpx.Timeout(settings.llm_timeout, connect=10.0),
        )

    async def chat(self, question, system_prompt=None, temperature=0.7):
        try:
            resp = await self._call_with_retry(messages, temperature)
        except (APITimeoutError, APIConnectionError):
            raise LLMError("LLM 调用超时或连接失败")
        except RateLimitError:
            raise LLMError("LLM 限流：请稍后再试")
        except APIError as exc:
            raise LLMError(f"LLM 上游错误：{exc.message}")

        answer = self._extract_content(resp)
        if not answer.strip():
            raise LLMError("LLM 返回了空内容")

        return {"answer": answer, "model": resp.model, "latency_ms": latency_ms}

注意它 raise 的是 LLMError（业务异常），不是 HTTPException。

注意：铁律：Service 不认识 HTTP。不 import FastAPI、不 import Request/Response。它不知道自己是被谁调用的。

Layer 3 — `db/` 数据层

只做一件事：和数据库说话。

python

def init_conversations(db_path=None):
    """启动时调一次，幂等建表."""
    with _connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                id TEXT PRIMARY KEY,
                title TEXT, user_id TEXT,
                created_at TEXT, updated_at TEXT
            )
        """)

纯 SQL，没有业务判断。换数据库只需改这个文件。

注意：铁律：不判业务规则、不调 LLM、不依赖 FastAPI。

4. 完整请求的"生命旅程"

Code

POST /api/v1/chat  { "question": "你好" }
  │
  ▼
main.py：TraceMiddleware 生成 trace_id，匹配路由
  │
  ▼
api/chat.py：Pydantic 校验 → 调 llm_service.chat(...)
  │                                    ↑ 跨层边界
  ▼
services/llm_service.py：构造 messages → 调 OpenAI → 提取结果 → return
  │
  ▼
api/chat.py：ChatResponse(**result) → JSON 序列化
  │
  ▼
200 OK  { "answer": "...", "latency_ms": 380 }

如果 LLM 调用失败：

Code

llm_service.chat() raise LLMError("超时")
  → api/chat.py 不 try/except，异常向上冒泡
  → main.py @app.exception_handler(AppError) 捕获
  → 返回 502 + { "code": "LLM_ERROR", "message": "..." }

关键：路由层完全不写 try/except。错误处理集中在 main.py 一个地方。

5. 反面教材：不分层的代价

python

# 路由层包打天下（能跑但活不过 3 个月）
@router.post("/chat")
async def chat_bad(req: dict):
    if "question" not in req:
        return {"error": "missing question"}, 400

    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    try:
        resp = client.chat.completions.create(model="gpt-4", messages=[...])
    except Exception as e:
        return {"error": str(e)}, 500

    conn = sqlite3.connect("app.db")
    conn.execute("INSERT INTO ...", (...))
    conn.commit()

    return {"answer": resp.choices[0].message.content}

分层前 vs 分层后：

想做的事	不分层	分层后
加日志	每个路由写一遍	service 一处
换数据库	全项目搜 `sqlite3` 改 30 处	`db/` 一处
加 RBAC	每个路由复制粘贴	`deps.py` + 签名
加重试	每个 OpenAI 调用包 `tenacity`	`llm_service` 一处
加流式	改 50 个路由	`llm_service` 加一个方法
加新接口	复制整段代码	写路由 + 复用 service

6. 课程版 vs 生产版的数据层

	课程版（函数式）	生产版（Repository OOP）
数据库	SQLite + 标准库	PostgreSQL + SQLAlchemy
写法	`def get_conv(id) -> dict`	`class ConvRepo: def get(self, id)`
学习曲线	平缓	需要 DI + ORM 知识
测试	monkeypatch	mock Repository
适合	教学、原型、小项目	生产、多人协作

本质一样：都是把"数据访问"封装得可测试、可替换。只是封装方式不同。

7. 分层心法 5 条

路由层薄如纸 — 超过 20 行就是味道不对
Service 不识 HTTP — 可被 HTTP、CLI、定时任务、MCP 调用，完全无感
数据层只写 SQL — 不判业务、不调 LLM
异常分两种 — 业务异常（AppError 子类）由 service 抛 → 全局 handler 翻译成 HTTP；系统异常（bug）让它崩
Depends 是"换零件"的能力 — 测试时一行代码替换登录用户

8. 自测：5 个问题

问题 1：agent_chat 函数只有 4 行，遵守了哪几条"分层心法"？

问题 2：为什么权限检查用 Depends(require_permission(...)) 而不是在函数体里 if not user.has_permission(...)？

问题 3："所有接口打统一访问日志"应该在哪一层做？为什么不在每个路由函数里写？

问题 4：把 SQLite 换成 PostgreSQL，要改哪些目录？哪些绝不能动？

问题 5（最重要）：services/some_service.py 里出现 raise HTTPException(status_code=400)，错在哪？怎么改？

答案与解析

问题 1：agent_chat 的 4 行藏着 4 条心法

python

async def agent_chat(req: AgentChatRequest) -> AgentChatResponse:
    if not req.question.strip():
        raise ValidationError("question 不能只是空白字符")
    return await agent_service.chat(req.question, temperature=req.temperature)

路由层薄如纸：4 行，业务全在 agent_service.chat
不写 try/except：ValidationError 向上冒泡，全局 handler 兜底
不调 OpenAI / 不写 SQL：调的只有 service
Pydantic 接管输入校验：函数只补 Pydantic 兜不住的语义校验

最明显的一条：路由完全不知道 LLM 会不会调工具。业务决策藏在 service 里。

问题 2：Depends vs 函数体内检查

维度	`Depends(require_permission)`	`if not user.has_permission`
可读性	签名即文档，5 秒看懂接口权限	必须读函数体
可测试	`app.dependency_overrides` 一行替换	需要 mock 多层
可组合	横向叠 5 个依赖只变签名	函数体变 50 行"检查地狱"

核心认知：Depends 不是"另一种权限写法"，是 FastAPI 给的横切关注点（cross-cutting）注入机制。

问题 3：统一访问日志在中间件层

答案：在 TraceMiddleware，不在路由函数里。

三个理由：

DRY：50 个路由 = 50 处复制，格式一改全得改
职责单一：路由只管 HTTP → service → HTTP，打日志是基础设施的事
完整生命周期：中间件能拿到请求开始时间和结束时间

分层决策速查：

关注点	放哪	例子
所有请求都要	Middleware	trace_id、CORS、限流
一组接口共享	Depends	RBAC、租户隔离
单个接口特有	签名 Depends	特定权限
业务规则触发	service 内部	余额不足、订单已支付

问题 4：换数据库的改动范围

该改的：app/db/（SQL 方言 + 连接池）、app/core/config.py（加配置项）、requirements.txt（加驱动）

绝对不能动的：app/api/、app/services/、app/tools/、app/prompts/

这就是分层的"红利兑现时刻"：改 1 层，其他 4 层岿然不动。

问题 5：service 里 raise HTTPException 错在哪

表面错：Service 不应该 import FastAPI。违反了"Service 不识 HTTP"。

深层错（3 个连锁问题）：

失去复用：CLI/定时任务调这个 service → import HTTPException → 状态码 400 在这些场景毫无意义
业务语义泄漏："question 为空"是业务事实，不是 HTTP 事实。400 只是 HTTP 通道的翻译结果
测试变复杂：测 service 要 pytest.raises(HTTPException)，还要检查 status_code

正确写法：

python

# services/some_service.py
from app.core.exceptions import ValidationError

async def do_something(question: str):
    if not question:
        raise ValidationError("question 不能为空")

然后 main.py 的全局 handler 自动把 AppError 子类翻译成 HTTP 响应：

code

Service: raise ValidationError("question 不能为空")  ← 业务事实
   ↓
Router: 不 try/except，向上冒泡
   ↓
main.py: @app.exception_handler(AppError) 兜住
   ↓
HTTP: 400 { "code": "VALIDATION_ERROR", "message": "..." }

3 层各司其职：service 抛业务异常 → 路由放行 → 全局 handler 翻译成 HTTP。

本节要点

分层是把"变化频率不同的事"放到不同文件——HTTP、业务、SQL、外部服务各管各的
路由 ≤ 20 行，Service 不识 HTTP，数据层只写 SQL
业务异常（AppError）由 service 抛，全局 handler 翻译——service 永远不碰 HTTP 状态码
Depends + 全局异常 handler 是两大横切武器——权限、日志、错误翻译统一收编
换数据库只改 db/ 和配置——分层让改动范围可控，这就是工程价值

第 2 课：FastAPI / Service 分层 —— 工程项目的脊椎骨

1. 为什么必须分层

2. 实际项目目录结构

3. 每一层的"职责圣旨"

Layer 1 — api/ 路由层

Layer 2 — services/ 服务层

Layer 3 — db/ 数据层

4. 完整请求的"生命旅程"

5. 反面教材：不分层的代价

6. 课程版 vs 生产版的数据层

7. 分层心法 5 条

8. 自测：5 个问题

答案与解析

问题 1：agent_chat 的 4 行藏着 4 条心法

问题 2：Depends vs 函数体内检查

问题 3：统一访问日志在中间件层

问题 4：换数据库的改动范围

问题 5：service 里 raise HTTPException 错在哪

本节要点

Layer 1 — `api/` 路由层

Layer 2 — `services/` 服务层

Layer 3 — `db/` 数据层