第 4 周 练习骨架（只读展示）¶

"""第 4 周 · 练习 1 · FastAPI 入门：路径参数、查询参数、请求体、response_model

学习目标：用 FastAPI 写出 5 个端点；理解 200/201/204/404/422 分别在什么时候出现；
      用 Pydantic 模型做输入校验（`NoteIn`）与输出裁剪（`response_model=NoteOut`）；
      用 `TestClient` 测 API；打开 `/docs` 手动调一次，看 `/openapi.json` 长什么样。

用法：uv run pytest exercises/week4/w4_01_fastapi_basics -q
      uv run fastapi dev exercises/week4/w4_01_fastapi_basics/app.py
      （起服务后开 http://127.0.0.1:8000/docs 手动调一次）
做法：把每个 TODO 换成你的实现，反复运行测试直到 10 个全绿。
      存储就是模块级的字典 `NOTES`，不要引入数据库；`reset_storage()` 已给出，
      测试会调用它。

要点回顾：
- 路径参数写在路径里（`/notes/{note_id}`）并在函数签名里声明类型，FastAPI 自动转换 +
校验；
- 查询参数用 `Annotated[int, Query(ge=1, le=50)]` 这种写法加约束，违反约束自动 422；
- 请求体用 Pydantic 模型接收，字段约束写在模型里（`Field(min_length=1)`）；
- `response_model=NoteOut` 决定"响应里出现哪些字段"——漏掉它就可能泄漏内部字段；
- 找不到资源用 `raise HTTPException(status_code=404, detail="...")`，不要 `return
None`；
- 删除成功返回 204，且响应体必须为空（函数 `return None`）。
"""

from typing import Annotated

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI(title="Notes API（第 4 周练习 1）", version="0.1.0")


# ---- 传输模型（已给出，不用改） ----
class NoteIn(BaseModel):
    """客户端发来的笔记：标题必填且非空，正文与标签可省略。

    例：NoteIn(title="读书笔记", tags=["python"]).body -> ""
    """

    title: Annotated[str, Field(min_length=1, max_length=100)]
    body: str = ""
    tags: list[str] = Field(default_factory=list)


class NoteOut(BaseModel):
    """返回给客户端的笔记：比 NoteIn 多一个 id。

    例：NoteOut(id=1, title="t", body="", tags=[]).id -> 1
    """

    id: int
    title: str
    body: str
    tags: list[str]


# ---- 内存存储（已给出，不用改） ----
NOTES: dict[int, NoteOut] = {}
_next_id = 1


def _new_id() -> int:
    """返回下一个笔记 id（从 1 开始自增）。

    例：空存储时第一次调用 -> 1，第二次 -> 2
    """
    global _next_id
    note_id = _next_id
    _next_id += 1
    return note_id


def reset_storage() -> None:
    """清空内存存储并把自增 id 归位（测试的 fixture 会调用它）。

    例：reset_storage() 之后 NOTES == {} 且下一个 id 是 1
    """
    global _next_id
    NOTES.clear()
    _next_id = 1


# ---- 5 个端点：函数体要你写 ----
@app.get("/health")
def health() -> dict[str, str]:
    """健康检查：固定返回 {"status": "ok"}。

    例：GET /health -> 200 {"status": "ok"}
    """
    # TODO: 在这里实现
    raise NotImplementedError


@app.get("/notes", response_model=list[NoteOut])
def list_notes(
    tag: str | None = None,
    limit: Annotated[int, Query(ge=1, le=50)] = 10,
) -> list[NoteOut]:
    """按 id 升序列出笔记；给了 tag 就只留标签里含它的；最多返回 limit 条。

    例：GET /notes?tag=python&limit=2 -> 200 [最多 2 条带 python 标签的笔记]
        GET /notes?limit=0 -> 422（Query(ge=1) 帮你挡掉）
    """
    # TODO: 在这里实现（提示：sorted(NOTES) 拿到升序的 id，切片用 [:limit]）
    raise NotImplementedError


@app.get("/notes/{note_id}", response_model=NoteOut)
def get_note(note_id: int) -> NoteOut:
    """按 id 取一条笔记；不存在时 raise HTTPException(404, detail="note 3 not found")。

    例：GET /notes/1 -> 200 {"id": 1, ...}；GET /notes/999 -> 404
    """
    # TODO: 在这里实现
    raise NotImplementedError


@app.post("/notes", response_model=NoteOut, status_code=201)
def create_note(payload: NoteIn) -> NoteOut:
    """新建笔记：分配 id、存进 NOTES、返回 201 与新笔记。

    例：POST /notes {"title": "t"} -> 201 {"id": 1, "title": "t", "body": "", "tags":
    []}
        POST /notes {"title": ""} -> 422（NoteIn 的 min_length=1 帮你挡掉）
    """
    # TODO: 在这里实现（提示：_new_id() 与 NoteOut(id=..., **payload.model_dump())）
    raise NotImplementedError


@app.delete("/notes/{note_id}", status_code=204)
def delete_note(note_id: int) -> None:
    """删除一条笔记：成功返回 204 空响应；不存在时 404。

    例：DELETE /notes/1 -> 204 且响应体为空；再 DELETE /notes/1 -> 404
    """
    # TODO: 在这里实现（注意：函数不要 return 任何值）
    raise NotImplementedError

"""第 4 周 · 练习 1 的测试（已给出，不要改）：10 个 TestClient 用例。

用法：uv run pytest exercises/week4/w4_01_fastapi_basics -q
读法：先读测试再写实现——测试就是需求说明书。
"""

from collections.abc import Iterator

import pytest
from app import app, reset_storage
from fastapi.testclient import TestClient


@pytest.fixture
def client() -> Iterator[TestClient]:
    """每个测试一个干净的存储 + 一个 TestClient。"""
    reset_storage()
    with TestClient(app) as test_client:
        yield test_client
    reset_storage()


def _create(client: TestClient, title: str, **extra: object) -> dict[str, object]:
    """辅助函数：POST 一条笔记并返回响应体。"""
    response = client.post("/notes", json={"title": title, **extra})
    assert response.status_code == 201, response.text
    return response.json()


def test_health_returns_ok(client: TestClient) -> None:
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}


def test_openapi_json_lists_the_routes(client: TestClient) -> None:
    response = client.get("/openapi.json")
    assert response.status_code == 200
    paths = response.json()["paths"]
    assert "/health" in paths
    assert "/notes" in paths
    assert "/notes/{note_id}" in paths
    assert "post" in paths["/notes"]
    assert "delete" in paths["/notes/{note_id}"]


def test_create_note_returns_201_with_id(client: TestClient) -> None:
    # 顺手验证 response_model 的裁剪作用：多传的 secret 不能出现在响应里
    body = _create(
        client,
        "第一条笔记",
        body="正文",
        tags=["python", "web"],
        secret="不该出现在响应里",
    )
    assert body["id"] == 1
    assert body["title"] == "第一条笔记"
    assert body["body"] == "正文"
    assert body["tags"] == ["python", "web"]
    assert set(body) == {"id", "title", "body", "tags"}


def test_create_note_rejects_empty_title(client: TestClient) -> None:
    response = client.post("/notes", json={"title": ""})
    assert response.status_code == 422
    assert response.json()["detail"][0]["loc"] == ["body", "title"]


def test_get_note_by_id(client: TestClient) -> None:
    created = _create(client, "读书笔记")
    response = client.get(f"/notes/{created['id']}")
    assert response.status_code == 200
    assert response.json() == created


def test_get_missing_note_returns_404(client: TestClient) -> None:
    response = client.get("/notes/999")
    assert response.status_code == 404
    assert "999" in response.json()["detail"]


def test_list_notes_respects_limit(client: TestClient) -> None:
    for index in range(3):
        _create(client, f"笔记 {index}")
    response = client.get("/notes", params={"limit": 2})
    assert response.status_code == 200
    titles = [item["title"] for item in response.json()]
    assert titles == ["笔记 0", "笔记 1"]


def test_list_notes_filters_by_tag(client: TestClient) -> None:
    _create(client, "A", tags=["python"])
    _create(client, "B", tags=["web"])
    _create(client, "C", tags=["python", "web"])
    response = client.get("/notes", params={"tag": "python"})
    assert response.status_code == 200
    assert [item["title"] for item in response.json()] == ["A", "C"]


def test_list_notes_rejects_limit_out_of_range(client: TestClient) -> None:
    assert client.get("/notes", params={"limit": 0}).status_code == 422
    assert client.get("/notes", params={"limit": 51}).status_code == 422


def test_delete_note_returns_204_then_404(client: TestClient) -> None:
    created = _create(client, "待删除")
    deleted = client.delete(f"/notes/{created['id']}")
    assert deleted.status_code == 204
    assert deleted.content == b""
    assert client.delete(f"/notes/{created['id']}").status_code == 404
    assert client.get("/notes").json() == []

"""第 4 周 · 练习 2 · 分层后的 notes 应用包。

分层：routers（协议层，只做 HTTP ↔ 模型转换）→ repository（存储，Protocol + 内存实现），
      schemas（Pydantic 传输模型）、errors（领域异常）、deps（依赖注入）、main（组装）。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
"""

"""依赖注入（已给出，不用改）：设置与仓储从这里取。

要点：`Depends` 只是"构造函数传依赖"的框架版；测试里用
      `app.dependency_overrides[get_repo] = lambda: InMemoryNoteRepository()`
      换掉真实现，
      用完记得 `app.dependency_overrides.clear()`（忘了会污染下一个测试）。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
"""

from functools import lru_cache
from typing import Annotated

from fastapi import Depends, Request
from pydantic_settings import BaseSettings, SettingsConfigDict

from notes_app.repository import NoteRepository


class Settings(BaseSettings):
    """应用设置：环境变量 NOTES_APP_NAME / NOTES_MAX_PAGE_SIZE 可覆盖默认值。

    例：Settings(max_page_size=2).max_page_size -> 2
    """

    model_config = SettingsConfigDict(env_prefix="NOTES_", extra="ignore")

    app_name: str = "notes-app"
    max_page_size: int = 50


@lru_cache
def get_settings() -> Settings:
    """返回全局唯一的设置对象（lru_cache = 模块级单例的正经写法）。

    例：get_settings() is get_settings() -> True
    """
    return Settings()


def get_repo(request: Request) -> NoteRepository:
    """返回 lifespan 在启动时挂到 app.state 上的仓储。

    例：测试里用 dependency_overrides 把它换成 InMemoryNoteRepository 实例
    """
    repo: NoteRepository = request.app.state.repo
    return repo


SettingsDep = Annotated[Settings, Depends(get_settings)]
RepoDep = Annotated[NoteRepository, Depends(get_repo)]

"""领域异常层次（已给出，不用改）：这个文件绝不 import fastapi。

为什么：存储层/服务层只知道"没找到""重名了"，把它翻译成哪个 HTTP 状态码是协议层的事。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
"""


class NoteError(Exception):
    """笔记领域所有异常的基类。

    例：isinstance(NoteNotFoundError(1), NoteError) -> True
    """


class NoteNotFoundError(NoteError):
    """按 id 找不到笔记（协议层映射成 404）。

    例：str(NoteNotFoundError(3)) -> 'note 3 not found'
    """

    def __init__(self, note_id: int) -> None:
        """记下找不到的 id，方便日志与响应体里带上它。"""
        super().__init__(f"note {note_id} not found")
        self.note_id = note_id


class NoteConflictError(NoteError):
    """标题重复（协议层映射成 409）。

    例：str(NoteConflictError("读书笔记")) -> "note titled '读书笔记' already exists"
    """

    def __init__(self, title: str) -> None:
        """记下冲突的标题。"""
        super().__init__(f"note titled {title!r} already exists")
        self.title = title

"""组装层：create_app（工厂）、lifespan、异常处理器、/health。

用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
      做完后在文件末尾加一行 `app = create_app()`，就能
      `uv run uvicorn notes_app.main:app --reload`（在本目录下运行）。
做法：lifespan、两个异常处理器、health、create_app 的函数体是 TODO。
      为什么用工厂函数而不是模块级 `app = FastAPI()`：每个测试可以拿到一个全新的 app，
      互不干扰；`dependency_overrides` 也不会串味。
"""

from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from notes_app.deps import RepoDep, SettingsDep
from notes_app.errors import NoteConflictError, NoteNotFoundError
from notes_app.repository import InMemoryNoteRepository, NoteRepository
from notes_app.routers import notes

TITLE = "Notes API（第 4 周练习 2）"
ROUTERS = [notes.router]


def build_repo() -> NoteRepository:
    """建一个仓储实例（已给出）：真实项目里这里换成 SQLite 实现，其他代码不用动。

    例：build_repo().count() -> 0
    """
    return InMemoryNoteRepository()


def error_response(status_code: int, exc: Exception) -> JSONResponse:
    """把领域异常包成 {"detail": 消息} 的 JSON 响应（已给出）。

    例：error_response(404, NoteNotFoundError(3)).status_code -> 404
    """
    return JSONResponse(status_code=status_code, content={"detail": str(exc)})


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """启动时把仓储挂到 app.state.repo，关闭时清掉（替代已弃用的 on_event）。

    例：`with TestClient(create_app()) as c:` 进入时建仓储，退出时清理
    """
    # TODO: 在这里实现"启动"部分：app.state.repo = build_repo()
    raise NotImplementedError
    yield
    # TODO: 在这里实现"关闭"部分：del app.state.repo（真实项目里是关连接池）


def health(settings: SettingsDep, repo: RepoDep) -> dict[str, object]:
    """健康检查：返回 {"status": "ok", "app": 应用名, "notes": 当前笔记数}。

    例：GET /health -> 200 {"status": "ok", "app": "notes-app", "notes": 0}
    """
    # TODO: 在这里实现（settings.app_name 与 repo.count()）
    raise NotImplementedError


async def note_not_found_handler(request: Request, exc: Exception) -> JSONResponse:
    """把领域异常 NoteNotFoundError 映射成 404 {"detail": 异常消息}。

    例：NoteNotFoundError(3) -> 404 {"detail": "note 3 not found"}
    """
    # TODO: 在这里实现（一行：return error_response(404, exc)）
    raise NotImplementedError


async def note_conflict_handler(request: Request, exc: Exception) -> JSONResponse:
    """把领域异常 NoteConflictError 映射成 409 {"detail": 异常消息}。

    例：NoteConflictError("t") -> 409 {"detail": "note titled 't' already exists"}
    """
    # TODO: 在这里实现
    raise NotImplementedError


# 领域异常 → 处理器的映射表（已给出）：注册表模式就是"装饰器/字典 + 循环"
EXCEPTION_HANDLERS: dict[
    type[Exception], Callable[[Request, Exception], Awaitable[JSONResponse]]
] = {
    NoteNotFoundError: note_not_found_handler,
    NoteConflictError: note_conflict_handler,
}


def create_app() -> FastAPI:
    """应用工厂：建 FastAPI 实例、注册异常处理器、挂 /health 与 notes 路由。

    步骤：
    1. app = FastAPI(title=TITLE, lifespan=lifespan)
    2. 遍历 EXCEPTION_HANDLERS，逐个 app.add_exception_handler(exc_type, handler)
    3. app.add_api_route("/health", health, methods=["GET"], tags=["ops"])
    4. 遍历 ROUTERS 逐个 app.include_router(router)
    5. return app

    例：create_app().title == TITLE -> True
    """
    # TODO: 在这里实现
    raise NotImplementedError

"""领域对象 + 仓储接口 + 内存实现（已给出，不用改）：这个文件也不 import fastapi。

要点：`NoteRepository` 是 typing.Protocol——"谁长这样谁就能用"，不需要继承。
      路由只依赖这个协议，测试时塞一个内存实现，线上塞 SQLite 实现，代码不变。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
"""

from dataclasses import dataclass, field
from typing import Protocol

from notes_app.errors import NoteConflictError, NoteNotFoundError


@dataclass(slots=True)
class Note:
    """领域模型：一条笔记（普通 dataclass，不带任何 Web 概念）。

    例：Note(1, "读书笔记", "", ["python"]).title -> '读书笔记'
    """

    id: int
    title: str
    body: str
    tags: list[str] = field(default_factory=list)


class NoteRepository(Protocol):
    """笔记存储的接口：找不到抛 NoteNotFoundError，重名抛 NoteConflictError。

    例：InMemoryNoteRepository() 就是它的一个实现（无需继承）
    """

    def add(self, title: str, body: str, tags: list[str]) -> Note:
        """新增一条笔记并返回它（带上分配好的 id）。"""
        ...

    def get(self, note_id: int) -> Note:
        """按 id 取笔记；不存在时抛 NoteNotFoundError。"""
        ...

    def list(self, tag: str | None = None, limit: int = 10) -> list[Note]:
        """按 id 升序列出笔记，可按标签过滤，最多 limit 条。"""
        ...

    def update(
        self,
        note_id: int,
        title: str | None = None,
        body: str | None = None,
        tags: list[str] | None = None,
    ) -> Note:
        """局部更新：只改传进来的字段，返回更新后的笔记。"""
        ...

    def delete(self, note_id: int) -> None:
        """按 id 删除；不存在时抛 NoteNotFoundError。"""
        ...

    def count(self) -> int:
        """当前笔记条数。"""
        ...


class InMemoryNoteRepository:
    """内存实现：一个 dict 加一个自增 id，测试用它就够了。

    例：repo = InMemoryNoteRepository()；repo.add("t", "", []).id -> 1
    """

    def __init__(self) -> None:
        """建一个空存储。"""
        self._notes: dict[int, Note] = {}
        self._next_id = 1

    def add(self, title: str, body: str, tags: list[str]) -> Note:
        """新增一条笔记；标题已存在时抛 NoteConflictError。

        例：repo.add("读书笔记", "", []) 两次 -> 第二次抛 NoteConflictError
        """
        if any(note.title == title for note in self._notes.values()):
            raise NoteConflictError(title)
        note = Note(id=self._next_id, title=title, body=body, tags=list(tags))
        self._notes[note.id] = note
        self._next_id += 1
        return note

    def get(self, note_id: int) -> Note:
        """按 id 取笔记；不存在时抛 NoteNotFoundError。

        例：空仓储 repo.get(7) -> raise NoteNotFoundError
        """
        try:
            return self._notes[note_id]
        except KeyError:
            raise NoteNotFoundError(note_id) from None

    def list(self, tag: str | None = None, limit: int = 10) -> list[Note]:
        """按 id 升序列出笔记，可按标签过滤，最多 limit 条。

        例：repo.list(tag="python", limit=2) -> 最多 2 条带 python 标签的笔记
        """
        notes = [self._notes[note_id] for note_id in sorted(self._notes)]
        if tag is not None:
            notes = [note for note in notes if tag in note.tags]
        return notes[:limit]

    def update(
        self,
        note_id: int,
        title: str | None = None,
        body: str | None = None,
        tags: list[str] | None = None,
    ) -> Note:
        """局部更新；改标题撞上别人的标题时抛 NoteConflictError。

        例：repo.update(1, body="新正文").body -> '新正文'
        """
        note = self.get(note_id)
        if title is not None and title != note.title:
            if any(
                other.title == title
                for other in self._notes.values()
                if other.id != note_id
            ):
                raise NoteConflictError(title)
            note.title = title
        if body is not None:
            note.body = body
        if tags is not None:
            note.tags = list(tags)
        return note

    def delete(self, note_id: int) -> None:
        """按 id 删除；不存在时抛 NoteNotFoundError。

        例：repo.delete(1) 之后 repo.count() 少 1
        """
        if self._notes.pop(note_id, None) is None:
            raise NoteNotFoundError(note_id)

    def count(self) -> int:
        """当前笔记条数。

        例：空仓储 repo.count() -> 0
        """
        return len(self._notes)

"""路由包：每个资源一个模块，由 main.create_app() 用 include_router 挂上去。"""

"""notes 路由（协议层）：只做 HTTP ↔ 模型转换，业务规则在仓储/服务层。

规则：这里**不写** try/except 把领域异常翻译成 HTTPException——那是 main.py 里
      异常处理器的活。路由让 NoteNotFoundError / NoteConflictError 直接往上抛。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
做法：5 个路由函数的函数体是 TODO，签名与装饰器已给好，照 docstring 实现。
"""

from typing import Annotated

from fastapi import APIRouter, Query

from notes_app.deps import RepoDep, SettingsDep
from notes_app.repository import Note
from notes_app.schemas import NoteIn, NoteOut, NotePatch

router = APIRouter(prefix="/notes", tags=["notes"])


@router.get("", response_model=list[NoteOut])
def list_notes(
    repo: RepoDep,
    settings: SettingsDep,
    tag: str | None = None,
    limit: Annotated[int, Query(ge=1)] = 10,
) -> list[Note]:
    """列出笔记：可按 tag 过滤；limit 再大也不能超过 settings.max_page_size。

    例：GET /notes?tag=python&limit=100 且 max_page_size=50 -> 最多 50 条
    """
    # TODO: 在这里实现（提示：min(limit, settings.max_page_size) 后交给 repo.list）
    raise NotImplementedError


@router.post("", response_model=NoteOut, status_code=201)
def create_note(payload: NoteIn, repo: RepoDep) -> Note:
    """新建笔记：201 + 新笔记；标题重复时让 NoteConflictError 抛出去（→ 409）。

    例：POST /notes {"title": "t"} -> 201 {"id": 1, ...}
    """
    # TODO: 在这里实现（提示：repo.add(payload.title, payload.body, payload.tags)）
    raise NotImplementedError


@router.get("/{note_id}", response_model=NoteOut)
def get_note(note_id: int, repo: RepoDep) -> Note:
    """取一条笔记；不存在时让 NoteNotFoundError 抛出去（→ 404）。

    例：GET /notes/999 -> 404 {"detail": "note 999 not found"}
    """
    # TODO: 在这里实现（一行：return repo.get(note_id)）
    raise NotImplementedError


@router.patch("/{note_id}", response_model=NoteOut)
def patch_note(note_id: int, payload: NotePatch, repo: RepoDep) -> Note:
    """局部更新：只改请求里给了的字段，返回更新后的笔记。

    例：PATCH /notes/1 {"body": "新正文"} -> 200，title 不变
    """
    # TODO: 在这里实现（提示：把 payload 的三个字段传给 repo.update）
    raise NotImplementedError


@router.delete("/{note_id}", status_code=204)
def delete_note(note_id: int, repo: RepoDep) -> None:
    """删除笔记：204 空响应；不存在时 NoteNotFoundError（→ 404）。

    例：DELETE /notes/1 -> 204；再删一次 -> 404
    """
    # TODO: 在这里实现（注意不要 return 任何值）
    raise NotImplementedError

"""传输模型（已给出，不用改）：只描述"线上传什么"，和存储里的领域对象分开。

要点：NoteOut 开了 from_attributes=True，所以路由可以直接 return 领域对象 Note，
      由 response_model 负责按字段名取属性并裁剪；这就是"边界处映射"。
用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
"""

from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field


class NoteIn(BaseModel):
    """新建笔记的请求体：标题必填非空。

    例：NoteIn(title="读书笔记").tags -> []
    """

    title: Annotated[str, Field(min_length=1, max_length=100)]
    body: str = ""
    tags: list[str] = Field(default_factory=list)


class NotePatch(BaseModel):
    """局部更新的请求体：三个字段都可选，没传的字段保持原样。

    例：NotePatch(body="新正文").title is None -> True
    """

    title: Annotated[str | None, Field(default=None, min_length=1, max_length=100)]
    body: str | None = None
    tags: list[str] | None = None


class NoteOut(BaseModel):
    """响应体：从领域对象 Note 的属性读取。

    例：NoteOut.model_validate(Note(1, "t", "", [])).id -> 1
    """

    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    body: str
    tags: list[str]

"""第 4 周 · 练习 2 的测试（已给出，不要改）：依赖覆盖 + lifespan + 领域异常映射。

用法：uv run pytest exercises/week4/w4_02_fastapi_structure -q
读法：注意两种客户端——`plain_client` 不覆盖任何依赖（走 lifespan 建的仓储），
      `client` 用 dependency_overrides 塞进一个测试自己的内存仓储，退出时清理覆盖。
"""

from collections.abc import Iterator

import pytest
from fastapi.testclient import TestClient
from notes_app.deps import Settings, get_repo, get_settings
from notes_app.errors import NoteNotFoundError
from notes_app.main import create_app
from notes_app.repository import InMemoryNoteRepository


@pytest.fixture
def repo() -> InMemoryNoteRepository:
    """一个空的内存仓储，测试可以直接往里塞数据。"""
    return InMemoryNoteRepository()


@pytest.fixture
def plain_client() -> Iterator[TestClient]:
    """不覆盖依赖的客户端：进入上下文时会触发 lifespan。"""
    app = create_app()
    with TestClient(app) as test_client:
        yield test_client


@pytest.fixture
def client(repo: InMemoryNoteRepository) -> Iterator[TestClient]:
    """把 get_repo 换成上面的内存仓储；退出时清掉覆盖（忘了清就会污染别的测试）。"""
    app = create_app()
    app.dependency_overrides[get_repo] = lambda: repo
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()


def test_lifespan_puts_a_repo_on_app_state(plain_client: TestClient) -> None:
    response = plain_client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok", "app": "notes-app", "notes": 0}
    assert hasattr(plain_client.app.state, "repo")  # type: ignore[attr-defined]


def test_health_counts_notes_from_the_injected_repo(
    client: TestClient, repo: InMemoryNoteRepository
) -> None:
    repo.add("已存在的笔记", "", ["python"])
    body = client.get("/health").json()
    assert body["notes"] == 1


def test_create_then_get_note(client: TestClient) -> None:
    created = client.post("/notes", json={"title": "读书笔记", "tags": ["python"]})
    assert created.status_code == 201
    assert set(created.json()) == {"id", "title", "body", "tags"}
    note_id = created.json()["id"]
    fetched = client.get(f"/notes/{note_id}")
    assert fetched.status_code == 200
    assert fetched.json() == created.json()


def test_duplicate_title_maps_to_409(client: TestClient) -> None:
    client.post("/notes", json={"title": "重复"})
    response = client.post("/notes", json={"title": "重复"})
    assert response.status_code == 409
    assert "already exists" in response.json()["detail"]


def test_missing_note_404_comes_from_domain_error(
    client: TestClient, repo: InMemoryNoteRepository
) -> None:
    # 仓储抛的是领域异常，和 HTTP 无关
    with pytest.raises(NoteNotFoundError):
        repo.get(999)
    # 协议层把它映射成 404，detail 就是异常消息
    response = client.get("/notes/999")
    assert response.status_code == 404
    assert response.json() == {"detail": "note 999 not found"}


def test_patch_only_changes_given_fields(client: TestClient) -> None:
    note_id = client.post("/notes", json={"title": "原标题", "body": "原正文"}).json()[
        "id"
    ]
    patched = client.patch(f"/notes/{note_id}", json={"body": "新正文"})
    assert patched.status_code == 200
    assert patched.json()["title"] == "原标题"
    assert patched.json()["body"] == "新正文"
    assert client.patch("/notes/999", json={"body": "x"}).status_code == 404


def test_delete_returns_204_then_404(client: TestClient) -> None:
    note_id = client.post("/notes", json={"title": "待删除"}).json()["id"]
    deleted = client.delete(f"/notes/{note_id}")
    assert deleted.status_code == 204
    assert deleted.content == b""
    assert client.delete(f"/notes/{note_id}").status_code == 404


def test_list_filters_by_tag(client: TestClient, repo: InMemoryNoteRepository) -> None:
    repo.add("A", "", ["python"])
    repo.add("B", "", ["web"])
    repo.add("C", "", ["python", "web"])
    titles = [item["title"] for item in client.get("/notes?tag=python").json()]
    assert titles == ["A", "C"]


def test_limit_is_capped_by_settings(
    repo: InMemoryNoteRepository,
) -> None:
    app = create_app()
    app.dependency_overrides[get_repo] = lambda: repo
    app.dependency_overrides[get_settings] = lambda: Settings(max_page_size=2)
    for index in range(5):
        repo.add(f"笔记 {index}", "", [])
    with TestClient(app) as test_client:
        assert len(test_client.get("/notes?limit=100").json()) == 2
    app.dependency_overrides.clear()


def test_openapi_uses_router_prefix_and_tags(plain_client: TestClient) -> None:
    schema = plain_client.get("/openapi.json").json()
    assert "/notes" in schema["paths"]
    assert "/notes/{note_id}" in schema["paths"]
    assert schema["paths"]["/notes"]["get"]["tags"] == ["notes"]
    assert "patch" in schema["paths"]["/notes/{note_id}"]

"""练习 3 的包标记文件。

为什么需要它：练习 1 里已经有一个 `app.py` 了。pytest 默认的导入模式下，两个同名顶层模块
会撞车（先导入的那个赢），所以这里加 `__init__.py` 让本目录成为包，
测试里就写 `from w4_03_fastapi_async_sse.app import app`——名字唯一，谁也不影响谁。
"""

"""第 4 周 · 练习 3 · 异步端点、后台任务、SSE、中间件、API key

学习目标：分清 `def` 端点与 `async def` 端点（异步端点里绝不能有阻塞调用）；
      用 `BackgroundTasks` 在响应之后干活；用 `fastapi.sse` 推 SSE 事件流；
      写一个请求日志中间件；用 `Depends` 读 `X-API-Key` 做最小鉴权；
      用 `httpx.AsyncClient(transport=ASGITransport(app=app))` 写异步测试。

用法：uv run pytest exercises/week4/w4_03_fastapi_async_sse -q
      uv run fastapi dev exercises/week4/w4_03_fastapi_async_sse/app.py
做法：把每个 TODO 换成你的实现，反复运行测试直到 10 个全绿。
      已给出的部分（Settings、ImportRequest、reset_state、new_id、两个 slow 端点）
      不用改。

要点：
- `/slow-sync` 是 `def`，Starlette 把它丢线程池，阻塞的 time.sleep 不会卡住事件循环；
  `/slow-async` 是 `async def`，里面只能 `await asyncio.sleep`——写 time.sleep
  会卡死服务；
- SSE（FastAPI 0.135+）：端点函数**本身**是异步生成器，装饰器写
  `response_class=EventSourceResponse`，逐条 `yield ServerSentEvent(...)`；
  每条之间都要有 `await`，否则客户端断开时无法取消；
- 中间件里 `response = await call_next(request)` 前后各取一次时间就是耗时；
- API key 缺失是 401（没认证），错误是 403（认证了但没权限）。
"""

import asyncio
import time
import uuid
from collections.abc import AsyncIterator, Awaitable, Callable
from functools import lru_cache
from typing import Annotated

from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    Header,
    HTTPException,
    Query,
    Request,
    Response,
)
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict

app = FastAPI(title="Async & SSE Notes（第 4 周练习 3）", version="0.1.0")


class Settings(BaseSettings):
    """设置：环境变量 W4_API_KEY / W4_TICK_SECONDS / W4_SLOW_SECONDS 可覆盖。

    例：Settings().api_key -> 'dev-key'
    """

    model_config = SettingsConfigDict(env_prefix="W4_", extra="ignore")

    api_key: str = "dev-key"
    tick_seconds: float = 0.05
    slow_seconds: float = 0.2


@lru_cache
def get_settings() -> Settings:
    """返回全局唯一的设置对象。

    例：get_settings() is get_settings() -> True
    """
    return Settings()


SettingsDep = Annotated[Settings, Depends(get_settings)]

REQUEST_LOG: list[dict[str, object]] = []
JOBS: dict[str, dict[str, object]] = {}


class ImportRequest(BaseModel):
    """导入请求：假装要导入 files 个文件。

    例：ImportRequest(files=3).files -> 3
    """

    files: Annotated[int, Field(ge=1, le=100)]


def reset_state() -> None:
    """清空请求日志与任务表（测试的 fixture 会调用它）。

    例：reset_state() 之后 JOBS == {} 且 REQUEST_LOG == []
    """
    REQUEST_LOG.clear()
    JOBS.clear()


def new_id() -> str:
    """生成 8 位随机 id：请求 id 与 job id 都用它。

    例：len(new_id()) -> 8
    """
    return uuid.uuid4().hex[:8]


def verify_api_key(
    settings: SettingsDep,
    x_api_key: Annotated[str | None, Header()] = None,
) -> str:
    """校验 X-API-Key 头：没带 401，带错 403，正确则返回它。

    例：headers={"X-API-Key": "dev-key"} -> 'dev-key'
    """
    # TODO: 在这里实现（注意：key 只能来自 settings，绝不能硬编码在代码里）
    raise NotImplementedError


@app.middleware("http")
async def log_requests(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    """请求日志中间件：给每个请求一个 id，记录耗时与状态码，并写进响应头。

    例：任何响应都带 X-Request-ID 与 X-Process-Time 两个头
    """
    # TODO: 在这里实现
    #   1. request_id = new_id()；started = time.perf_counter()
    #   2. response = await call_next(request)；算出 elapsed
    #   3. 往 REQUEST_LOG 追加一条 dict：request_id / method / path / status / elapsed
    # 4. 给响应加两个头：X-Request-ID、X-Process-Time（f"{elapsed:.4f}"），并 return
    # response
    raise NotImplementedError


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

    例：GET /health -> 200 {"status": "ok"}
    """
    # TODO: 在这里实现
    raise NotImplementedError


@app.get("/events/countdown", response_class=EventSourceResponse)
async def countdown(
    settings: SettingsDep,
    n: Annotated[int, Query(ge=1, le=20)] = 5,
) -> AsyncIterator[ServerSentEvent]:
    """SSE 端点（0.135+ 写法）：端点函数本身是异步生成器，逐条 yield 事件。

    例：GET /events/countdown?n=3 -> 3 条 event=tick + 1 条 event=done
        线上格式：'event: tick' / 'data: {"remaining": 3}' / 空行
    """
    # TODO: 在这里实现（n 条 tick：先 await asyncio.sleep(settings.tick_seconds) 再
    #   yield ServerSentEvent(data={"remaining": remaining}, event="tick")；
    #   循环结束后 yield 一条 data={"total": n}、event="done"）
    raise NotImplementedError
    yield ServerSentEvent(data={"remaining": n}, event="tick")


async def run_import(job_id: str, files: int) -> None:
    """后台任务：逐个"导入"文件并更新 JOBS 里的进度。

    例：run_import("abc", 2) 结束后 JOBS["abc"]["status"] == 'done'
    """
    # TODO: 在这里实现（把 JOBS[job_id] 的 status 改成 "running"，
    #   逐个文件 await asyncio.sleep(0) 并更新 processed，最后 status 改成 "done"）
    raise NotImplementedError


@app.post("/import", status_code=202)
async def start_import(
    payload: ImportRequest, background_tasks: BackgroundTasks
) -> dict[str, str]:
    """收下导入请求，登记任务并立刻返回 202；真正的导入在响应之后跑。

    例：POST /import {"files": 3} -> 202 {"job_id": "...", "status": "queued"}
    """
    # TODO: 在这里实现（用 new_id() 生成 job_id，登记 status="queued"，
    # background_tasks.add_task(run_import, job_id, payload.files)，返回 job_id 与状态）
    raise NotImplementedError


@app.get("/jobs/{job_id}")
async def get_job(job_id: str) -> dict[str, object]:
    """查任务进度；没有这个任务时 404。

    例：GET /jobs/abc -> {"status": "done", "processed": 3, "total": 3}
    """
    job = JOBS.get(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail=f"job {job_id} not found")
    return job


@app.get("/slow-sync")
def slow_sync(settings: SettingsDep) -> dict[str, str]:
    """同步端点：阻塞 sleep 后返回。Starlette 会把它放线程池里跑。

    例：GET /slow-sync -> 200 {"kind": "sync"}
    """
    time.sleep(settings.slow_seconds)
    return {"kind": "sync"}


@app.get("/slow-async")
async def slow_async(settings: SettingsDep) -> dict[str, str]:
    """异步端点：await asyncio.sleep 后返回，5 个并发请求总耗时约等于一次。

    例：GET /slow-async -> 200 {"kind": "async"}
    """
    await asyncio.sleep(settings.slow_seconds)
    return {"kind": "async"}


@app.get("/admin/stats", dependencies=[Depends(verify_api_key)])
async def admin_stats() -> dict[str, int]:
    """受 X-API-Key 保护的统计端点。

    例：GET /admin/stats + 正确 key -> 200 {"jobs": 0, "requests": 1}
    """
    # TODO: 在这里实现（返回 JOBS 与 REQUEST_LOG 的条数）
    raise NotImplementedError