第 4 周周测答案（默认折叠）¶

闭卷做完、自己判完再看

点击展开

第 4 周周测答案¶

第 1–6 题已在 FastAPI 0.141 / Pydantic 2.13 上实际运行核对。

一、预测输出¶

201 {'title': 'a', 'tags': ['x'], 'id': 1} —— status_code=201 与 response_model 决定了状态码和输出字段。
422 [['body', 'title']] missing —— 请求体缺必填字段，FastAPI 用 Pydantic 的错误结构返回 detail 列表，loc 指出位置。
422 404 {'detail': 'not found'} —— 路径参数声明为 int，"abc" 转换失败是 422；存在但找不到资源是我们自己抛的 404。
422 200 —— Query(ge=1, le=100) 校验失败 422；"5" 字符串会被转成 int。
404 然后 200 —— dependency_overrides 把 get_db 换成返回空字典的函数，于是查不到；清理后恢复真实依赖。
['/notes', '/notes/{note_id}'] ['title'] —— OpenAPI 由路由与 Pydantic 模型自动生成，tags 有默认值所以不在 required。
/a：同步端点在线程池里跑，5 个请求并行，≈0.2 秒；/b：async def 里调 time.sleep 会阻塞事件循环，5 个请求排队，≈1.0 秒；/c：真正的异步等待，≈0.2 秒。这是本周最重要的一题。
Content-Type: text/event-stream；正文按 SSE 格式一行一个字段、事件之间空行：
```
event: tick
data: 0

event: tick
data: 1
```
抛 pydantic_core.ValidationError：api_key 是必填且没有来源（"Field required"）；debug 会被解析成 True（"1" 是真值）。缺配置在启动时立刻失败，比运行到一半才发现好。
状态码 404、JSON {"detail": "没有 x"}。好处：服务层/领域层不 import FastAPI，可以在 CLI、测试、脚本里复用；HTTP 状态码的映射只写一处（异常处理器），换框架或加 gRPC 接口时领域代码不用改。

二、找 bug¶

async def 端点里直接调 2 秒的同步查询会卡死事件循环，期间所有请求都停。改成 def report(...)（跑线程池），或 rows = await asyncio.to_thread(db.heavy_sql_query)，或改用异步驱动。
服务层 import 了 HTTPException，把"传输协议"混进业务：不能在非 HTTP 场景复用，测试也要 import FastAPI。抛领域异常 NoteNotFoundError，在 main.py 用 @app.exception_handler 映射为 404。
allow_origins=["*"] 与 allow_credentials=True 同时开是标准禁止的组合（浏览器会拒绝带凭证的通配请求，而且意味着任何网站都能带着用户 Cookie 调你的 API）。列白名单 allow_origins=["https://你的前端域名"]。
覆盖依赖后没清理，状态泄漏到其他测试。用 fixture：yield 之前设置、之后 app.dependency_overrides.clear()（或只 pop 自己设的那一项）。

on_event 已弃用。用 lifespan：

@asynccontextmanager
async def lifespan(app):
    app.state.db = connect()
    yield
    app.state.db.close()
app = FastAPI(lifespan=lifespan)

好处是启动和关闭写在同一个函数里，资源一定成对释放。

三、写代码（参考）¶

16.

from typing import Annotated
from fastapi import APIRouter, Depends, Response, status
from notebook.api.deps import get_repo
from notebook.repositories import NoteRepository
from notebook.schemas import NoteOut
from notebook.domain import NoteNotFoundError

router = APIRouter(prefix="/notes", tags=["notes"])
Repo = Annotated[NoteRepository, Depends(get_repo)]

@router.get("/{note_id}", response_model=NoteOut)
def read(note_id: int, repo: Repo):
    note = repo.get(note_id)
    if note is None:
        raise NoteNotFoundError(note_id)      # 由 main.py 的处理器映射为 404
    return NoteOut.from_domain(note)

@router.delete("/{note_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete(note_id: int, repo: Repo):
    if not repo.delete(note_id):
        raise NoteNotFoundError(note_id)
    return Response(status_code=204)

include_router 写在 main.py 的 create_app() 里：app.include_router(notes.router)，所有路由文件只定义 router，不碰 app。 17.

import sqlite3
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    app.state.conn = sqlite3.connect(settings.db_path, check_same_thread=False)
    migrate(app.state.conn)
    yield
    app.state.conn.close()

def get_repo(request: Request) -> SqliteNoteRepository:
    return SqliteNoteRepository(request.app.state.conn)

18.

from typing import Annotated
from fastapi import Depends, Header, HTTPException

def verify_api_key(
    x_api_key: Annotated[str | None, Header()] = None,
    settings: Settings = Depends(get_settings),
) -> None:
    if x_api_key is None:
        raise HTTPException(401, detail="缺少 X-API-Key")
    if x_api_key != settings.api_key:
        raise HTTPException(403, detail="API key 无效")

# 路由：@router.post("", dependencies=[Depends(verify_api_key)])
def test_missing_key(client):
    assert client.post("/notes", json={"title": "t"}).status_code == 401

def test_wrong_key(client):
    assert client.post("/notes", json={"title": "t"}, headers={"X-API-Key": "nope"}).status_code == 403

19.

import asyncio, json
from fastapi.sse import EventSourceResponse, ServerSentEvent

@router.get("/{job_id}/events")
async def job_events(job_id: str, jobs: Jobs):
    queue = jobs.subscribe(job_id)          # asyncio.Queue[dict | None]

    async def gen():
        while True:
            event = await queue.get()       # 挂起等待，不占 CPU
            if event is None:
                yield ServerSentEvent(data="{}", event="done")
                return
            yield ServerSentEvent(data=json.dumps(event, ensure_ascii=False), event="progress")

    return EventSourceResponse(gen())

必须 await：生成器跑在事件循环里，忙等（while not queue.empty(): pass）会独占循环，其他请求全部停摆，而且客户端断开时的取消也只有在 await 处才能被传递进来。 20.

api/routers/*   → 只 import：schemas、services、api/deps、domain 的异常类型；禁止 sqlite3/httpx 具体实现
api/deps.py     → 组装：settings、repositories 的具体实现 → 交给 services
services/       → 只 import：domain、repositories 的 Protocol；禁止 fastapi、sqlite3、httpx
repositories/   → Protocol（无依赖）+ 实现（import sqlite3）；禁止 fastapi、services
schemas/        → Pydantic 传输模型；只 import domain（做 from_domain/to_domain 映射）
domain/         → dataclass + 异常；不 import 任何框架，也不 import 上面任何一层
main.py         → create_app：lifespan、include_router、exception_handler（NotFound→404、Conflict→409、Invalid→422）、中间件

落地方式：领域异常类都继承 NotebookError 放在 domain/errors.py；服务层只 raise 它们；main.py 对每个子类注册 @app.exception_handler 返回对应状态码与 {"detail": ...}；测试里用 grep/脚本断言 services/、domain/ 里搜不到 fastapi、sqlite3、httpx。