HTTP、JSON 与 API 客户端

网络调用是失败率远高于本地函数调用的边界。可靠客户端必须设置超时、区分状态码和连接错误、验证响应格式，并谨慎决定是否重试。

一、学习目标

使用标准库发送 JSON 请求
设置超时和必要请求头
区分 HTTP 错误、网络错误和数据错误
设计可测试的客户端边界
理解幂等性与重试

二、发送 JSON 请求

import json
from urllib.request import Request, urlopen


def push_task(url: str, task: Task, token: str | None = None) -> dict[str, object]:
    payload = json.dumps(
        {
            "id": task.id,
            "title": task.title,
            "status": task.status.value,
            "updated_at": task.updated_at.isoformat(),
        }
    ).encode("utf-8")

    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    request = Request(url, data=payload, method="POST", headers=headers)
    with urlopen(request, timeout=10) as response:
        result = json.load(response)
    if not isinstance(result, dict):
        raise RuntimeError("服务器响应不是 JSON 对象")
    return result

任何网络请求都应有有限超时。默认无限等待会让 CLI 或工作线程永久卡住。

三、错误分类

from urllib.error import HTTPError, URLError

try:
    with urlopen(request, timeout=10) as response:
        result = json.load(response)
except HTTPError as exc:
    raise RuntimeError(f"服务器返回 HTTP {exc.code}") from exc
except URLError as exc:
    raise RuntimeError(f"无法连接同步服务：{exc.reason}") from exc
except json.JSONDecodeError as exc:
    raise RuntimeError("服务器返回了无效 JSON") from exc

HTTPError：已经收到 HTTP 响应，但状态码表示失败。
URLError：DNS、连接、TLS 等传输失败。
JSONDecodeError：响应体不符合期望格式。
业务格式错误：JSON 合法但缺少字段或类型错误。

不同错误对应不同诊断和重试策略。

四、验证响应

from typing import TypedDict


class SyncResult(TypedDict):
    task_id: int
    accepted: bool


def parse_sync_result(value: object) -> SyncResult:
    if not isinstance(value, dict):
        raise ValueError("响应必须是对象")
    task_id = value.get("task_id")
    accepted = value.get("accepted")
    if not isinstance(task_id, int) or not isinstance(accepted, bool):
        raise ValueError("响应字段无效")
    return {"task_id": task_id, "accepted": accepted}

类型注解不能验证网络数据，必须在边界执行运行时检查。

五、URL 与查询参数

from urllib.parse import urlencode, urljoin

base_url = "https://api.example.com/"
endpoint = urljoin(base_url, "tasks")
query = urlencode({"status": "todo", "page": 1})
url = f"{endpoint}?{query}"

不要手工拼接未经编码的用户输入。还要谨慎使用 urljoin：如果第二个参数是绝对 URL，它会替换主机。只允许程序定义的相对路径，或额外验证最终主机。

六、认证与秘密

token = os.environ.get("STUDY_TASKS_API_TOKEN")

Token 不写入代码和公开配置。
日志不得打印 Authorization 头。
只通过 HTTPS 发送凭据。
区分“未配置”“失效”“权限不足”。
服务端返回 401/403 时，不应无意义重试。

七、重试与幂等

读取请求或带幂等键的写入，可以对暂时性错误进行有限重试，并加入指数退避和随机抖动。

不应自动重试：

参数校验失败的 4xx；
普通非幂等 POST；
已知业务拒绝；
没有上限的重试。

如果创建操作必须重试，客户端生成幂等键，服务端保存并识别重复请求。

八、可测试设计

把网络传输封装在小函数或客户端类中，业务服务只依赖协议：

from typing import Protocol


class TaskSyncClient(Protocol):
    def push(self, task: Task) -> SyncResult: ...

测试业务服务时传入内存假实现，不发送真实请求。客户端自身测试可以启动本地临时 HTTP 服务器，覆盖状态码、超时和无效 JSON。

不要在单元测试中依赖公网服务。

九、第三方客户端

标准库适合理解协议和减少依赖。真实项目需要连接池、同步/异步统一接口、细粒度超时或更友好的 API 时，可评估 HTTPX 等维护活跃的客户端。

引入第三方库前仍需设计相同的错误分类、响应验证、重试和秘密处理；库不会替你决定业务契约。

十、常见错误

没有设置超时。
遇到所有异常都重试。
日志记录完整 Token 或响应体。
认为 HTTP 200 就代表响应结构正确。
在数据库事务内等待远程请求。

十一、练习与自测

启动本地测试服务器，分别返回 200、500 和无效 JSON。
为同步请求增加幂等键头。
将传输层替换成协议，并为服务编写无网络单元测试。

自测：

HTTP 错误与连接错误的差异是什么？
为什么网络数据需要运行时验证？
哪些请求可以安全重试？

十二、官方资料

上一篇：SQLite、事务与数据访问｜下一篇：并发编程与 asyncio