[!info] Related notes

• 所属 MOC：Web 后端 · 标准库 · 学习路线
• 前置：错误处理 · JSON · 程序结构
• 后续：FastAPI · HTTP 客户端 · 认证与 JWT · 优雅关闭 · Docker 部署
• 对照：Go 标准库 HTTP 服务

使用标准库构建 HTTP 服务

这个实践为什么出现

会 print("hello") 或会调 FastAPI 路由，不等于会交付 HTTP 服务。工程最小闭环是：

• 明确方法和路径
• 稳定的 JSON 契约与错误码
• 不监听端口也能测业务和编解码；HTTP 适配层另做请求测试
• 为超时、鉴权、优雅关闭留扩展点

Python 标准库 http.server 足够完成教学与内部小工具闭环；生产系统再换 ASGI/WSGI 框架，但请求生命周期与分层思想不变。

[!abstract] 一句话理解 Handler 做协议适配（读请求、写状态码/头/body）；业务函数保持可测；用 ThreadingHTTPServer 提供线程化连接处理，单测直接调用业务与编解码辅助函数。

目标

实现一个本地 JSON 服务：

接口	行为
GET /healthz	200 {"status":"ok"}
POST /echo	回显 UTF-8 JSON 对象；要求 application/json 与单个 Content-Length
非法 JSON、长度字段或截断请求体	400
缺少长度 / body 超过 64 KiB / 媒体类型不符	411 / 413 / 415
未知路径	404

并满足：

• 业务逻辑可无网络单测
• handler 不塞复杂规则
• 能说明升级到 FastAPI/uvicorn 时多了什么

最小可运行示例

先把示例放进业务场景，再看代码：

场景：内部平台要一个可 curl 的健康检查与调试 echo

发布流水线需要探活；开发联调需要一个 echo 接口确认网关没有改 body。
服务必须：

1. 健康检查足够简单稳定
2. echo 校验 JSON
3. 以后要能加鉴权/日志而不重写业务

# mini_http_service.py
# 业务意图：最小可测 JSON HTTP 服务。
# 教学点：
# - 业务函数与 handler 分离；
# - Content-Length / Content-Type；
# - 错误映射为 HTTP 状态码；
# - 单测不监听端口。

from __future__ import annotations

import json
import math
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from urllib.parse import urlparse


def health_payload() -> dict[str, str]:
    # 业务：探活响应体。保持无依赖，便于测试与缓存策略讨论。
    return {"status": "ok"}


def reject_constant(value: str) -> None:
    raise ValueError("invalid json")


def finite_float(value: str) -> float:
    number = float(value)
    if not math.isfinite(number):
        raise ValueError("number out of range")
    return number


def echo_payload(raw: bytes) -> dict[str, Any]:
    # 业务：解析 JSON；失败抛 ValueError 让协议层映射 400。
    try:
        text = raw.decode("utf-8")
        data = json.loads(text, parse_constant=reject_constant, parse_float=finite_float)
    except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc:
        raise ValueError("invalid json") from exc
    if not isinstance(data, dict):
        raise ValueError("json object required")
    return {"echo": data}


def to_json_bytes(payload: dict[str, Any]) -> bytes:
    return json.dumps(payload, ensure_ascii=True, allow_nan=False).encode("utf-8")


class AppHandler(BaseHTTPRequestHandler):
    """协议适配层：只负责 HTTP 语义，不堆业务规则。"""

    max_body_bytes = 64 * 1024

    def setup(self) -> None:
        super().setup()
        self.connection.settimeout(5)  # socket 无活动超时，不是请求总 deadline

    def _write_json(self, status: int, payload: dict[str, Any]) -> None:
        body = to_json_bytes(payload)
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        path = urlparse(self.path).path
        if path == "/healthz":
            self._write_json(200, health_payload())
            return
        self._write_json(404, {"error": "not found"})

    def do_POST(self) -> None:
        path = urlparse(self.path).path
        if path != "/echo":
            self._write_json(404, {"error": "not found"})
            return
        if self.headers.get_all("Transfer-Encoding"):
            self._write_json(501, {"error": "transfer encoding unsupported"})
            return
        lengths = self.headers.get_all("Content-Length", [])
        if not lengths:
            self._write_json(411, {"error": "content length required"})
            return
        if len(lengths) != 1 or not lengths[0].isascii() or not lengths[0].isdigit():
            self._write_json(400, {"error": "invalid content length"})
            return
        try:
            length = int(lengths[0])
        except ValueError:
            self._write_json(400, {"error": "invalid content length"})
            return
        if length > self.max_body_bytes:
            self._write_json(413, {"error": "body too large"})
            return
        if self.headers.get_content_type() != "application/json":
            self._write_json(415, {"error": "application/json required"})
            return
        try:
            raw = self.rfile.read(length)
        except TimeoutError:
            self._write_json(408, {"error": "body read timeout"})
            return
        if len(raw) != length:
            self._write_json(400, {"error": "incomplete body"})
            return
        try:
            payload = echo_payload(raw)
        except ValueError as exc:
            self._write_json(400, {"error": str(exc)})
            return
        self._write_json(200, payload)

    def log_message(self, fmt: str, *args: object) -> None:
        # 演示安静；生产应接 logging。
        return


def create_server(host: str = "127.0.x.x", port: int = 8765) -> ThreadingHTTPServer:
    return ThreadingHTTPServer((host, port), AppHandler)


def main() -> None:
    with create_server() as server:
        print("listening on http://127.0.x.x:8765")
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass  # 离开 with 后关闭监听 socket；不是完整在途请求排空


if __name__ == "__main__":
    main()

建议运行：

# 终端 1
python mini_http_service.py

# 终端 2
curl -s http://127.0.x.x:8765/healthz
curl -s -X POST http://127.0.x.x:8765/echo \
  -H 'Content-Type: application/json' \
  -d '{"x":1}'
curl -s -X POST http://127.0.x.x:8765/echo \
  -H 'Content-Type: application/json' \
  -d '{'

期望输出：

{"status": "ok"}
{"echo": {"x": 1}}
{"error": "invalid json"}

无端口业务测试

这些测试只覆盖业务函数和编码，不覆盖 HTTP 状态、Header、路由或网络读取；后者应另用本地临时端口请求验证。

# test_mini_http_business.py
# 业务意图：验证健康检查与 echo 契约（pytest 风格，也可直接 assert 运行）。

from mini_http_service import echo_payload, health_payload, to_json_bytes


def test_health_payload() -> None:
    assert health_payload() == {"status": "ok"}


def test_echo_ok() -> None:
    assert echo_payload(b'{"x": 1}') == {"echo": {"x": 1}}


def test_echo_bad_json() -> None:
    try:
        echo_payload(b"{")
        raise AssertionError("expected ValueError")
    except ValueError as exc:
        assert "invalid json" in str(exc)


def test_json_bytes_encoding() -> None:
    body = to_json_bytes({"message": "你好"})
    assert body == b'{"message": "\\u4f60\\u597d"}'


if __name__ == "__main__":
    test_health_payload()
    test_echo_ok()
    test_echo_bad_json()
    test_json_bytes_encoding()
    print("all business tests passed")

建议运行：

# 与 mini_http_service.py 同目录
python test_mini_http_business.py
# 或：pytest -q test_mini_http_business.py

期望输出：

all business tests passed

结合场景再看四个关注点

1. handler 薄、业务纯：换 FastAPI 时多半只换适配层。
2. 状态码语义：400 客户端错，404 路径错，200 成功。
3. 单测不 bind 端口：CI 更快更稳。
4. 标准库服务的边界：路由、校验、OpenAPI、中间件生态弱，生产请上框架。

核心概念与准确模型

1. 请求生命周期（本示例）

sequenceDiagram
  participant C as Client
  participant S as ThreadingHTTPServer
  participant H as AppHandler
  participant B as Business funcs
  C->>S: TCP + HTTP request
  S->>H: do_GET / do_POST
  H->>B: health_payload / echo_payload
  B-->>H: dict or ValueError
  H-->>C: status + headers + JSON body

2. Handler 职责

该做	不该做
解析 path/method/headers/body	拼复杂 SQL/领域规则
映射异常 → 状态码	吞异常返回 200
写 Content-Type/Length	忘记编码与长度

3. ThreadingHTTPServer

• 每连接线程处理，适合教学与低并发工具
• 不是 ASGI；无原生 async 路由
• 生产常见：uvicorn/gunicorn + FastAPI/Starlette/Django

4. JSON 契约

• 本文显式处理的响应返回 JSON；未实现的 HTTP 方法等由基类返回 HTML 错误，不应宣称所有响应均为 JSON
• 响应使用 ASCII 转义输出非 ASCII 字符，客户端 JSON 解码后仍得到原值；避免孤立代理码点导致 UTF-8 编码失败。
• 生产错误体要控制信息泄漏

设计动机

1. 先掌握协议适配与可测业务，框架只是加速器。
2. 与 Go net/http 实践对齐：newMux/可测 handler 同一思想。
3. 给后续中间件/鉴权/关闭留缝：main 只组装。

边界情况与反直觉行为

1. 先写 body 再改状态码

send_response 与 send_header 先缓冲响应头，end_headers 将其写出；提交后不能再改状态。可能失败的解析和编码应在提交前完成。

2. Content-Length 与流式

本例只支持单个十进制 Content-Length、最多 64 KiB，拒绝 Transfer-Encoding。默认 HTTP/1.0 响应关闭连接；不要直接改成 keep-alive 而忽略错误响应后尚未消费的请求体。较大或流式请求应交给成熟服务器/框架。

3. 路径与查询串

self.path 可能含 ?query；应用 urlparse(self.path).path。

4. 线程模型

ThreadingHTTPServer 共享进程内状态时要线程安全；默认示例无共享可变全局。

5. 安全

本例绑定 loopback，并设置 body 大小上限和 socket 无活动超时；仍缺少鉴权、限流与完整服务器防护，http.server 不适合作为公网生产 API。

常见误区

[!warning] 常见误区：只会 curl，不写业务测试 错误理解：手测即质量。
正确模型：业务函数表驱动测试 + 少量集成 curl/httpx。

[!warning] 常见误区：业务全写在 do_POST 错误理解：一个方法打天下。
正确模型：协议层/应用层分离。

[!warning] 常见误区：标准库服务直接当生产 API 网关 错误理解：少依赖=更可靠。
正确模型：缺生态默认值（超时矩阵、校验、观测、进程模型）。

[!warning] 常见误区：所有错误都 500 错误理解：省事。
正确模型：4xx/5xx 语义是契约的一部分。

与相邻概念对比

方案	适用	差异
http.server（本篇）	教学、内部工具	路由/中间件弱
FastAPI/Starlette	现代 API	类型校验、OpenAPI、async
Django	全栈/管理后台	电池齐全，更重
Go net/http	同思想对照	语言与并发模型不同

工程实践

扩展路径（同一骨架）

1. POST 下单：echo_payload 模式 → 校验字段 → 调 service
2. 鉴权：读 Authorization，映射 401/403 → 认证与 JWT
3. 日志：替换 log_message，接 日志 与 correlation id
4. 优雅关闭：shutdown() 必须从 serve_forever 所在线程之外调用，否则可能死锁；随后 server_close。ThreadingHTTPServer 的请求线程默认是 daemon，完整排空还需另行设计，见 优雅关闭
5. 容器：入口 python -m myapp → Docker
6. 换框架：业务函数原样迁到 FastAPI 路由依赖注入

生产前最低清单

• 示例已限制 64 KiB；部署层也应设置限制
• 超时与进程模型（多 worker）
• 健康检查与就绪检查分离（深探再议）
• HTTPS 终止位置明确
• 依赖与配置可复现

可验证实验

2026-10-07 使用 Python 3.12.3 运行文中业务测试，并在 loopback 临时端口验证状态码、长度、媒体类型、JSON、截断 body、超限和读超时。空 body 按非法 JSON 返回 400；非有限数值和浮点溢出也拒绝。未实现的方法仍使用基类的 501 HTML 响应。

实验 1：健康检查

curl -i /healthz 见 200 与 Content-Type: application/json。

实验 2：非法 JSON

POST { 得 400 与 error 字段。

实验 3：业务单测

运行 python test_mini_http_business.py 全绿。

实验 4：未知路径

GET /nope 得 404。

实验 5（扩展）：方法不允许

给 /healthz 发 POST，观察现状（已实现 do_POST，但非 /echo 路径返回 404）并思考应否返回 405。

本节总结

• 标准库 HTTP 服务用来建立生命周期与分层，不是与 FastAPI 对立。
• 可测业务 + 薄 handler 是跨语言通用工程习惯。
• 会测、会映射错误码、知道生产缺口，才算完成 0→1。

自测题

概念题

1. 为什么业务函数要与 do_GET/do_POST 分离？
2. ThreadingHTTPServer 与 ASGI 服务器核心差别？

代码推理题

3. echo_payload 收到 b"[1,2]" 应如何表现（按本示例）？

工程思考题

4. 若要把本服务换成 FastAPI，哪些文件/函数最可能原样保留？

参考答案

1. 便于无端口测试、替换协议层、避免 HTTP 细节污染领域规则。
2. ASGI 面向异步应用接口与中间件生态；http.server 是同步教学向服务器。
3. json.loads 得到 list，非 dict，抛 ValueError("json object required") → 400。
4. health_payload/echo_payload 及领域校验逻辑；handler 换成路由函数。

延伸阅读与资料来源

资料	类型	支撑内容
http.server	文档	标准库服务器
json	文档	编解码
WSGI	文档	另一类 Python Web 接口
ASGI spec	规范	异步服务接口
FastAPI	文档	生产向框架路径

笔记元信息

• 角色：howto / 可验证闭环
• 深度对齐：build-a-go-http-service-with-standard-library
• 下一步：中间件化日志、鉴权、优雅关闭、换 ASGI