python

《Python编程实战》18.2 迭代开发、联调与压测

把 TaskFlow 真的写出来：SQLAlchemy 2.1.4 模型、scrypt 加 JWT 认证、带缓存的 FastAPI 路由、后台任务队列，并用 pytest 跑出 11 例通过、86% 覆盖率，起 uvicorn 用 httpx 联调、自写压测器压出真实 QPS 与延迟分位。

Leeting Yan 6 分钟阅读 2808 字

本节目标：按垂直切片把 TaskFlow 一层层写通，用 pytest 验证、用 uvicorn + httpx 联调、用自写压测器测出真实吞吐与延迟，并让后台任务具备幂等、重试与死信能力。
适用版本：Python 3.12+（实测 3.14.6）；FastAPI 0.143.0、SQLAlchemy 2.1.4、pytest 9.1.1

18.2 迭代开发、联调与压测

上一节把图纸画好了，这一节真刀真枪把它写出来。所有代码都在本机 Python 3.14.6 上跑过，下面的测试输出、HTTP 响应、压测数字都是真实结果。

18.2.1 垂直切片：一次打通一层，而不是横着铺

新手容易「先把所有 model 写完、再写所有 schema、再写所有路由」——横着铺的坏处是中间没有一刻能跑。更好的节奏是垂直切片：数据层 → 认证层 → 路由层 → 任务层，每一刀切下去都能立刻验证。

切片交付物验证方式
1 数据层models.py + Alembic 迁移alembic upgrade head 建表成功
2 认证层security.py + /auth/*单元测试哈希与令牌往返
3 路由层routers/tickets.py + 缓存集成测试 CRUD 与隔离
4 任务层tasks.py脚本验证幂等/重试/死信

18.2.2 数据层：SQLAlchemy 2.1.4 的类型化模型

SQLAlchemy 2.x 的新风格是 DeclarativeBase + Mapped[] + mapped_column()，让 ORM 模型本身带上类型信息（第 6 章详述）。核心三张表：

class Ticket(Base):
    __tablename__ = "tickets"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(String(4000), default="")
    status: Mapped[TicketStatus] = mapped_column(default=TicketStatus.OPEN)
    priority: Mapped[Priority] = mapped_column(default=Priority.MEDIUM)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    owner: Mapped[User] = relationship(back_populates="tickets")
    comments: Mapped[list["Comment"]] = relationship(
        back_populates="ticket", cascade="all, delete-orphan"
    )

会话工厂用 async_sessionmaker，expire_on_commit=False 让提交后对象仍可读（否则异步下访问属性会触发隐式 IO 报错）：

engine = create_async_engine(settings.database_url, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

18.2.3 认证层：scrypt 存密码，HS256 签发令牌

认证拆成两半：存密码用标准库 hashlib.scrypt（慢哈希 + 每用户随机盐，第 9 章的结论），发令牌用 HS256 手写 JWT（本机无 PyJWT，标准库 hmac 足够）：

def hash_password(pw: str) -> str:
    salt = secrets.token_bytes(16)
    dk = hashlib.scrypt(pw.encode(), salt=salt, n=2**14, r=8, p=1, dklen=32)
    return f"scrypt$16384$8$1${salt.hex()}${dk.hex()}"

def create_token(sub: str, secret: str, ttl: int) -> str:
    header = _b64e(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
    now = int(time.time())
    payload = _b64e(json.dumps({"sub": sub, "iat": now, "exp": now + ttl}).encode())
    sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest()
    return f"{header}.{payload}.{_b64e(sig)}"

校验时用 hmac.compare_digest 比签名（防时序攻击），并显式检查 exp。当前用户通过依赖注入拿到：

async def get_current_user(session, creds) -> User:
    payload = decode_token(creds.credentials, settings.secret_key)
    user = await session.scalar(select(User).where(User.id == int(payload["sub"])))
    ...
CurrentUser = Annotated[User, Depends(get_current_user)]

18.2.4 路由层：缓存旁路与租户隔离

查单张工单走缓存旁路（cache-aside）：先查缓存，未命中查库并回填；任何写操作后失效缓存：

@router.get("/{ticket_id}", response_model=TicketOut)
async def get_ticket(ticket_id: int, session: SessionDep, user: CurrentUser) -> TicketOut:
    key = cache.ticket_key(ticket_id)
    cached = await cache.get_cached(key)
    if cached is not None:
        return TicketOut.model_validate(cached)
    ticket = await _get_owned(ticket_id, user.id, session)
    out = TicketOut.model_validate(ticket)
    await cache.set_cached(key, out.model_dump(mode="json"))
    return out

租户隔离收敛到一个函数里，所有读写都过它——这是 US-6 的落地点，也是安全的关键：

async def _get_owned(ticket_id: int, owner_id: int, session) -> Ticket:
    ticket = await session.get(Ticket, ticket_id)
    if ticket is None or ticket.owner_id != owner_id:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "工单不存在")
    return ticket

注意这里返回 404 而不是 403：对不属于自己的资源，不该暴露「它存在」这一信息。

18.2.5 测试：pytest 跑出 11 例通过

测试用 httpx.ASGITransport 把 app 内联进来，配一个临时 SQLite 文件库，测试之间互不干扰：

@pytest_asyncio.fixture
async def client(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/test.db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    maker = async_sessionmaker(engine, expire_on_commit=False)

    async def override():
        async with maker() as session:
            yield session

    app.dependency_overrides[get_session] = override
    await cache.redis.flushall()
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()
    await engine.dispose()

dependency_overrides 把生产会话换成测试会话，是 FastAPI 测试的标准手法。真实运行结果：

$ pytest -v
tests/test_auth.py::test_password_hash_roundtrip PASSED                  [  9%]
tests/test_auth.py::test_token_roundtrip PASSED                          [ 18%]
tests/test_auth.py::test_token_rejects_wrong_secret PASSED               [ 27%]
tests/test_auth.py::test_register_and_login PASSED                       [ 36%]
tests/test_auth.py::test_login_wrong_password PASSED                     [ 45%]
tests/test_tickets.py::test_create_and_get_ticket PASSED                 [ 54%]
tests/test_tickets.py::test_list_pagination PASSED                       [ 63%]
tests/test_tickets.py::test_update_invalidates_cache PASSED              [ 72%]
tests/test_tickets.py::test_cross_tenant_isolation PASSED                [ 81%]
tests/test_tickets.py::test_comment_flow PASSED                          [ 90%]
tests/test_tickets.py::test_requires_auth PASSED                         [100%]
============================== 11 passed in 1.15s ==============================

覆盖率（真实输出，pytest-cov）：

Name                      Stmts   Miss  Cover   Missing
-------------------------------------------------------
app/cache.py                 14      0   100%
app/models.py                41      0   100%
app/routers/auth.py          25      0   100%
app/routers/tickets.py       61      6    90%   61, 76, 85-88
app/security.py              38      3    92%   43-44, 51
app/tasks.py                 36     21    42%   18, 24-27, 31-49
app/main.py                  30      8    73%   17-23, 39
-------------------------------------------------------
TOTAL                       312     43    86%

tasks.py 只有 42%，因为它由独立脚本验证（见 18.2.8），不在 pytest 里。这提示一件事：覆盖率是地图不是终点——缺失行里既有真没测的（main.py 的 lifespan），也有「另有验证途径」的。

18.2.6 联调：起 uvicorn，用 httpx 打真实 HTTP

单元测试跑在内联传输上，看不到真实网络行为。联调阶段必须起真服务：uvicorn 监听 8018，httpx 发真实 HTTP。真实输出：

=== 联调 ===
healthz -> {'status': 'ok'}
POST /tickets -> 201 X-Process-Time-ms=13.19 body={'id': 1, 'title': '登录页 500', 'body': '', 'status': 'open', 'priority': 'high', 'owner_id': 1, 'created_at': '2026-10-09T04:08:08'}
GET /tickets/1 (cache miss) -> 200
GET /tickets/1 (cache hit)  -> 200 X-Process-Time-ms=0.97
PATCH /tickets/1 -> 200 status=resolved
GET /tickets -> total=1 items=1
POST /tickets/1/comments -> 201 body=已复现
GET comments (未实现) -> 405

读三件事：缓存命中把延迟从 13 ms 压到 0.97 ms（中间件把每请求耗时写进 X-Process-Time-ms 响应头，这是第 3 章的可观测性落地）；PATCH 后状态真的变成 resolved（说明缓存失效生效，否则会读到旧值）；GET comments 返回 405 而不是 404，因为该路径只注册了 POST——这种「意料之外的错误码」正是联调要抓的。

18.2.7 压测：自写压测器，别用眼睛估

本机没有 locust，用 concurrent.futures + time.perf_counter 自写一个最小压测器，httpx 并发打真实服务：

def hit(_: int) -> float:
    t0 = time.perf_counter()
    r = httpx.get(f"{BASE}/tickets?limit=5", headers=headers, timeout=10, trust_env=False)
    dt = (time.perf_counter() - t0) * 1000
    if r.status_code != 200:
        errors += 1
    return dt

start = time.perf_counter()
with cf.ThreadPoolExecutor(max_workers=concurrency) as pool:
    latencies = list(pool.map(hit, range(total)))
elapsed = time.perf_counter() - start

真实压测结果（16 并发 × 800 请求）：

=== 压测 ===
并发=16 总请求=800 用时=2.27s
QPS=352.5 错误=0
P50=42.9ms P95=64.1ms P99=92.5ms max=106.3ms mean=45.1ms

别只看 QPS：P99 是 92.5 ms，是 P50 的两倍多，说明尾部延迟被拉长——这类请求很可能落在缓存未命中或 SQLite 写锁上。压测的意义不是得到一个好看的数字，而是逼出「平均值掩盖的尾部」。这里的瓶颈是 SQLite 单写者模型 + scrypt 登录（每次约 36 ms，但压测走的是已登录的读接口，不重复登录），所以 QPS 350 属于本地环境合理量级；真上 PostgreSQL + 连接池会显著不同。

18.2.8 后台任务：幂等、重试、死信

创建工单时要投递一条通知任务。任务队列用 asyncio + fakeredis 实现最小可用版（本机无 Celery/RabbitMQ，机制完全一致）：

async def enqueue(kind: str, payload: dict, job_id: str | None = None) -> str:
    job_id = job_id or str(uuid.uuid4())
    if await redis.sadd(DEDUP, job_id) == 0:   # 幂等：同一 job_id 只入队一次
        return job_id
    await redis.rpush(QUEUE, json.dumps({"id": job_id, "kind": kind, "payload": payload}))
    return job_id

worker 消费失败时重试最多 3 次，仍失败则进死信队列（DLQ）而不是丢弃：

while attempts < 3:
    try:
        await _handle(job["kind"], job["payload"])
        break
    except Exception as exc:
        attempts += 1
        if attempts >= 3:
            await redis.rpush(DLQ, json.dumps({**job, "error": str(exc), "attempts": attempts}))

真实运行结果：

入队 job-1 -> job-1
重复入队 job-1 -> job-1 (同 id，被幂等去重)
队列长度: 2
worker 处理任务数: 2
队列剩余: 0
死信队列: [{'id': 'job-2', 'kind': 'boom', 'payload': {'ticket_id': 2}, 'error': '未知任务类型: boom', 'attempts': 3}]

三个机制各就各位：幂等（重复 job-1 没进队列，长度仍是 2）、重试（boom 任务被试了 3 次）、死信（失败任务带着 error 与 attempts 落进 DLQ，可人工排查而非静默丢失）。这正是第 8 章「定时任务、幂等与死信处理」的完整落地。

延伸阅读

小结

  • 开发节奏用垂直切片，每切一层都能立刻验证，避免「半天跑不起来」。
  • SQLAlchemy 2.1.4 用 Mapped[] + async_sessionmaker(expire_on_commit=False)，异步下才不踩隐式 IO 的坑。
  • 认证存密码用 scrypt、发令牌用 HS256；租户隔离收敛到一个函数，越权一律返回 404。
  • 缓存旁路：读先查缓存、写后失效；实测命中把单请求从 13 ms 压到 0.97 ms。
  • 测试用 ASGITransport + dependency_overrides，11 例通过、86% 覆盖率；覆盖率是地图不是终点。
  • 联调必须起真服务；压测看 P99 不看均值；后台任务要同时具备幂等、重试、死信三件套。

代码跑通了，但它还只是「本地能跑」。下一节处理从「能跑」到「敢上线」之间的距离：配置与密钥怎么管、迁移怎么回滚、健康检查怎么写、上线后怎么复盘、系统往哪演进。

阅读导航：上一节：需求拆解与架构设计 · 下一节：上线、复盘与后续演进 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时