第 4 周周测（闭卷，40 分钟，20 题）¶

不开 REPL、不开 AI。答案见 exercises/solutions/week4/quiz_week4_answers.md。≥16 分通过。

前 6 题共用这段服务：

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

class NoteIn(BaseModel):
    title: str
    tags: list[str] = []

class NoteOut(NoteIn):
    id: int

DB: dict[int, NoteOut] = {}
def get_db() -> dict[int, NoteOut]:
    return DB

@app.post("/notes", response_model=NoteOut, status_code=201)
def create(note: NoteIn, db: Annotated[dict, Depends(get_db)]):
    out = NoteOut(id=len(db) + 1, **note.model_dump())
    db[out.id] = out
    return out

@app.get("/notes/{note_id}", response_model=NoteOut)
def read(note_id: int, db: Annotated[dict, Depends(get_db)]):
    if note_id not in db:
        raise HTTPException(404, detail="not found")
    return db[note_id]

@app.get("/notes")
def list_notes(limit: Annotated[int, Query(ge=1, le=100)] = 10):
    return list(DB.values())[:limit]

client = TestClient(app)

一、预测输出（10 题）¶

python r = client.post("/notes", json={"title": "a", "tags": ["x"]}) print(r.status_code, r.json())
python r = client.post("/notes", json={"tags": ["x"]}) print(r.status_code, [e["loc"] for e in r.json()["detail"]], r.json()["detail"][0]["type"])
python print(client.get("/notes/abc").status_code, client.get("/notes/99").status_code, client.get("/notes/99").json())
python print(client.get("/notes", params={"limit": 0}).status_code, client.get("/notes", params={"limit": "5"}).status_code)
python app.dependency_overrides[get_db] = lambda: {} print(client.get("/notes/1").status_code) # 上面第 1 题已经创建了 id=1 app.dependency_overrides.clear() print(client.get("/notes/1").status_code)
python spec = client.get("/openapi.json").json() print(sorted(spec["paths"]), spec["components"]["schemas"]["NoteIn"]["required"])

下面两个端点各被 5 个客户端同时请求，总耗时分别大约多少？

@app.get("/a")
def a():
    time.sleep(0.2); return 1
@app.get("/b")
async def b():
    time.sleep(0.2); return 1
@app.get("/c")
async def c():
    await asyncio.sleep(0.2); return 1

python from fastapi.sse import EventSourceResponse, ServerSentEvent @app.get("/ticks") async def ticks(): async def gen(): for i in range(2): yield ServerSentEvent(data=str(i), event="tick") yield ServerSentEvent(data="bye", event="done") return EventSourceResponse(gen()) 客户端收到的响应 Content-Type 是什么？正文前两行长什么样？
python class Settings(BaseSettings): model_config = SettingsConfigDict(env_prefix="NB_", env_file=".env") api_key: str debug: bool = False 环境里有 NB_DEBUG=1、没有 NB_API_KEY、也没有 .env：Settings() 会怎样？
python class NotFound(Exception): ... @app.exception_handler(NotFound) async def _(request, exc): return JSONResponse(status_code=404, content={"detail": str(exc)}) @app.get("/x") def x(): raise NotFound("没有 x") client.get("/x") 的状态码与 JSON 是什么？服务层抛的是 NotFound 而不是 HTTPException，这有什么好处？

二、找 bug（5 题）¶

python @app.get("/report") async def report(db: Annotated[Repo, Depends(get_repo)]): rows = db.heavy_sql_query() # 同步 sqlite3，要跑 2 秒 return rows
python # services/note_service.py from fastapi import HTTPException class NoteService: def get(self, note_id): note = self.repo.get(note_id) if note is None: raise HTTPException(404)
python app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True)
python def test_read(client): app.dependency_overrides[get_repo] = lambda: FakeRepo() assert client.get("/notes/1").status_code == 200 # 没有清理 overrides，下一个测试文件里的集成测试莫名其妙地用上了 FakeRepo
python @app.on_event("startup") async def startup(): app.state.db = connect()

三、写代码（5 题）¶

写一个 APIRouter(prefix="/notes")，含 GET /notes/{id} 与 DELETE /notes/{id}（204），仓储通过 Annotated[NoteRepository, Depends(get_repo)] 注入；说明 include_router 放在哪里。
写 lifespan(app)：启动时创建 SQLite 连接放进 app.state，关闭时 close()；再写 get_repo 依赖从 request.app.state 取连接构造仓储。
写 verify_api_key(x_api_key: Annotated[str | None, Header()] = None, settings=Depends(get_settings))：缺头 401、错误 403；再写它的两个测试。
写 SSE 端点 GET /jobs/{job_id}/events：从 asyncio.Queue 读进度事件逐条 yield ServerSentEvent(...)，收到 None 哨兵发 event="done" 后结束；说明为什么生成器里必须 await 而不是忙等。
用文字画出 Notebook Service 的分层（routers / services / repositories / schemas / domain），标出每层允许 import 什么、禁止 import 什么，并说明"领域异常在协议层统一映射为 HTTP"这条规则如何落地。