使用标准库构建 HTTP 服务

用标准库实现可测试的最小 JSON HTTP 服务：handler 与业务分离、健康检查与 echo、无端口单测，并接到框架与优雅关闭路径。

#type / howto #status / growing #tech / dev #resource / python #tech / lang / python

[!info] 关联笔记

所属 MOC：Web 后端 · 标准库 · 学习路线

前置：错误处理 · JSON · 程序结构

后续：FastAPI · HTTP 客户端 · 认证与 JWT · 优雅关闭 · Docker 部署

对照：Go 标准库 HTTP 服务

使用标准库构建 HTTP 服务

这个实践为什么出现

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

明确方法和路径
稳定的 JSON 契约与错误码
不监听端口也能测 handler/业务
为超时、鉴权、优雅关闭留扩展点

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

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

目标

实现一个本地 JSON 服务：

接口	行为
`GET /healthz`	`200 {"status":"ok"}`
`POST /echo`	回显 JSON body
非法 JSON	`400`
未知路径	`404`

并满足：

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

最小可运行示例

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

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

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

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

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

from __future__ import annotations

import json
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 echo_payload(raw: bytes) -> dict[str, Any]:
    # 业务：解析 JSON；失败抛 ValueError 让协议层映射 400。
    try:
        text = raw.decode("utf-8") if raw else "{}"
        data = json.loads(text)
    except (UnicodeDecodeError, json.JSONDecodeError) 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=False).encode("utf-8")


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

    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
        length = int(self.headers.get("Content-Length", "0"))
        raw = self.rfile.read(length) if length > 0 else b""
        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:
    server = create_server()
    print("listening on http://127.0.x.x:8765")
    server.serve_forever()


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"}

无端口测试（与 Go httptest 同思想）

把业务函数单独测，不启动服务器：

# 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_content_type_shape() -> None:
    body = to_json_bytes({"status": "ok"})
    assert b"status" in body


if __name__ == "__main__":
    test_health_payload()
    test_echo_ok()
    test_echo_bad_json()
    test_json_bytes_content_type_shape()
    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

结合场景再看四个关注点

handler 薄、业务纯：换 FastAPI 时多半只换适配层。
状态码语义：400 客户端错，404 路径错，200 成功。
单测不 bind 端口：CI 更快更稳。
标准库服务的边界：路由、校验、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，便于客户端统一解析
ensure_ascii=False 方便中文调试
生产错误体要控制信息泄漏

设计动机

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

边界情况与反直觉行为

1. 先写 body 再改状态码

send_response/end_headers 后基本不能反悔；可能失败的解析要放在写头之前（本示例已如此）。

2. Content-Length 与流式

小 JSON 可一次写；大文件要流式与分块策略。

3. 路径与查询串

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

4. 线程模型

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

5. 安全

echo 类接口切勿对公网裸奔；缺鉴权、限流、体大小限制。

常见误区

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

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

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

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

与相邻概念对比

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

工程实践

扩展路径（同一骨架）

POST 下单：echo_payload 模式 → 校验字段 → 调 service
鉴权：读 Authorization，映射 401/403 → 认证与 JWT
日志：替换 log_message，接日志与 correlation id
优雅关闭：信号 + server.shutdown() → 优雅关闭
容器：入口 python -m myapp → Docker
换框架：业务函数原样迁到 FastAPI 路由依赖注入

生产前最低清单

请求体大小上限
超时与进程模型（多 worker）
健康检查与就绪检查分离（深探再议）
HTTPS 终止位置明确
依赖与配置可复现

可验证实验

实验 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，观察现状（本示例 404/未实现 do_POST 分支）并思考应否返回 405。

本节总结

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

自测题

概念题

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

代码推理题

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

工程思考题

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

参考答案

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

延伸阅读与资料来源

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

笔记元信息

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