Course Builder个人学习工作台 · 本地模式
Modern Python：面向有经验开发者的回归路线

Capstone：组装并验证 TextLab

目标：能够把前面已经实现的 schema、异步分类、TaskGroup、FastAPI 与 SQLAlchemy repository 组合成可从干净环境运行和测试的服务。

最后一课不再引入新框架。你已经分别完成：

  • Pydantic request/response 边界；
  • FastAPI route 与 TestClient；
  • AsyncIO timeout 和 TaskGroup；
  • SQLAlchemy Engine、Session、事务和 repository；
  • uv、Ruff、Pyright、pytest 质量门。

现在只做依赖方向和生命周期的整合：HTTP 层依赖 service，service 依赖 provider 与 repository，外部实现通过构造或 FastAPI dependency 注入。

最终目录

textlab/
├── pyproject.toml
├── uv.lock
├── scripts/check.sh
├── src/textlab/
│   ├── __init__.py
│   ├── api.py
│   ├── db.py
│   ├── schemas.py
│   └── service.py
└── tests/
    ├── test_api.py
    ├── test_db.py
    └── test_service.py

服务层：只负责用例

建立 src/textlab/service.py：

from __future__ import annotations

import asyncio
from typing import Protocol

from textlab.schemas import MessageInput, MessageOutput


class AsyncTextProvider(Protocol):
    async def classify(self, text: str) -> str:
        raise NotImplementedError


class MessageWriter(Protocol):
    def add(self, *, text: str, category: str, priority: int) -> int:
        raise NotImplementedError


class LocalProvider:
    async def classify(self, text: str) -> str:
        await asyncio.sleep(0)
        return "refund" if "退款" in text else "general"


class TextService:
    def __init__(self, provider: AsyncTextProvider, writer: MessageWriter) -> None:
        self._provider = provider
        self._writer = writer

    async def process(self, message: MessageInput) -> tuple[int, MessageOutput]:
        async with asyncio.timeout(2):
            category = await self._provider.classify(message.text)

        output = MessageOutput(
            text=message.text,
            category=category,
            priority=message.priority,
        )
        message_id = await asyncio.to_thread(
            self._writer.add,
            text=output.text,
            category=output.category,
            priority=output.priority,
        )
        return message_id, output

这里 SQLite repository 是同步实现，因此用 to_thread 避免在 async route 中直接阻塞。真实项目也可以选择 SQLAlchemy async engine；那是另一条完整技术路线，不应在 Capstone 临时切换。

API 组装：依赖有唯一创建位置

在 api.py 中创建 Engine、session factory、repository 和 service，并用 lifespan 初始化表。route 只调用 service 并映射 timeout：

from __future__ import annotations

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session

from textlab.db import Base, MessageRepository, build_engine
from textlab.schemas import MessageInput, MessageOutput
from textlab.service import LocalProvider, TextService

engine = build_engine("sqlite+pysqlite:///./textlab.db")
repository = MessageRepository(lambda: Session(engine))
service = TextService(LocalProvider(), repository)


@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
    Base.metadata.create_all(engine)
    yield
    engine.dispose()


app = FastAPI(title="TextLab", version="1.0.0", lifespan=lifespan)


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/messages", response_model=MessageOutput, status_code=201)
async def create_message(message: MessageInput) -> MessageOutput:
    try:
        _, output = await service.process(message)
        return output
    except TimeoutError as exc:
        raise HTTPException(status_code=504, detail="processing timed out") from exc

这是完整的最终 api.py。它依赖第 15–17 课已经完整建立的 schemas.py、db.py、MessageRepository 和本课的 service.py；没有新的隐藏类型或省略实现。

可控的 service 测试

tests/test_service.py 使用 fake，不等待真实网络：

import asyncio

from textlab.schemas import MessageInput, Metadata
from textlab.service import TextService


class FakeProvider:
    async def classify(self, text: str) -> str:
        return "fake"


class FakeWriter:
    def __init__(self) -> None:
        self.rows: list[tuple[str, str, int]] = []

    def add(self, *, text: str, category: str, priority: int) -> int:
        self.rows.append((text, category, priority))
        return len(self.rows)


def test_process() -> None:
    writer = FakeWriter()
    service = TextService(FakeProvider(), writer)
    message = MessageInput(
        text="hello",
        priority=1,
        metadata=Metadata(source="test"),
    )

    message_id, output = asyncio.run(service.process(message))

    assert message_id == 1
    assert output.category == "fake"
    assert writer.rows == [("hello", "fake", 1)]

若你的 pytest 已配置 async plugin，也可写 async test；这里用 asyncio.run 保持依赖最少。

最终验收

从项目目录执行：

rm -rf .venv
uv sync --locked
./scripts/check.sh
uv run fastapi dev src/textlab/api.py

另一个终端请求 health 和 messages。预期：

  • clean sync 成功；
  • format、lint、type、tests 全部通过；
  • 合法 message 返回 201；
  • 缺失 metadata 或额外字段返回 422；
  • provider 超时测试返回 504；
  • repository 测试证明事务失败不会留下记录。

Windows 请使用等价删除命令和 PowerShell check script。

独立挑战：让两个 provider 共同工作

在不改 HTTP schema 的前提下，同时运行分类与摘要 provider。使用 TaskGroup，让任一失败时取消另一个；把 summary 加入内部持久化模型，若要加入公共响应则明确版本化 API。

验收标准与实现提示

在 TextService.process 中：

async with asyncio.timeout(2):
    async with asyncio.TaskGroup() as group:
        category_task = group.create_task(self._provider.classify(message.text))
        summary_task = group.create_task(self._summarizer.summarize(message.text))
    category = category_task.result()
    summary = summary_task.result()

测试至少覆盖：两者成功；classify 失败时 summarize 的 finally cleanup 被调用；整体 timeout；数据库写入只在两个结果成功后发生。不要用真实 sleep 竞争，使用 asyncio.Event 控制 fake。

3.12 完成后如何看 3.14

先保持项目测试在 3.12 通过，再增加 3.14 CI 矩阵。重点观察延迟标注语义、free-threaded build 的第三方扩展兼容性、ProcessPoolExecutor 启动方式和新的 interpreter 能力。升级不是把版本号改掉，而是让同一验收集证明行为仍成立。

参考资料