Skip to content

stage3-http-clients-tutorial.md

Latest commit

 

History

History
223 lines (151 loc) · 8.5 KB

stage3-http-clients-tutorial.md

File metadata and controls

223 lines (151 loc) · 8.5 KB
Raw

第 3 阶段：HTTP 客户端教程（requests / httpx）

0. 学习目标

这一阶段的目标是：能调用 LLM API、Agent 服务、内部 HTTP 接口，并理解同步与异步客户端的差异。

学完后，你应该能：

  • 用 requests 发同步 GET / POST，处理 headers、params、JSON body、timeout、status code。
  • 用 httpx.Client 封装一个可复用的同步 API client。
  • 用 httpx.AsyncClient 并发请求多个 URL。
  • 给 client 加 timeout、重试、错误日志。
  • 给 HTTP 代码写不依赖真网络的测试。

第 2 阶段学的是把语法组织成"能每天用的小工具"；第 3 阶段学的是让这些工具能和外部世界对话。

1. Java/Spring 对照速查

Java / Spring Python 说明
RestTemplate requests 同步、简单，最快上手
RestTemplate / 阻塞式 WebClient httpx.Client 可复用连接池的同步客户端
响应式 WebClient httpx.AsyncClient 非阻塞、并发
@Bean RestTemplate + 构造注入 把 client 作为参数注入 测试时换成 mock
Resilience4j Retry 手写 RetryPolicy（本阶段） 最小重试实现
try-with-resources (AutoCloseable) with / __enter__/__exit__ 自动释放连接
MockRestServiceServer httpx.MockTransport 离线模拟 HTTP

2. requests：最快的同步调用

requests 适合脚本里快速调一次接口。

import requests


def fetch_json(url: str) -> dict:
    response = requests.get(url, params={"page": 1}, timeout=30)
    response.raise_for_status()   # 非 2xx 抛 requests.HTTPError
    return response.json()

POST JSON：

response = requests.post(url, json={"name": "widget"}, timeout=30)

要点（务必记住）：

  • 一定写 timeout。requests 默认没有超时，网络一卡线程就无限挂起——这是最常见的生产事故。
  • response.raise_for_status() 把 4xx/5xx 变成异常，相当于 RestTemplate 非 2xx 抛 HttpStatusCodeException。
  • json=... 会自动设 Content-Type: application/json 并序列化；不要自己 json.dumps 再传 data=。
  • 多次请求同一主机时复用 requests.Session()，能复用 TCP 连接（类似复用 RestTemplate）。

3. httpx：同步 + 异步的现代选择

httpx 的 API 和 requests 几乎一样，但额外支持异步，是现代 Python 项目（含很多 Agent SDK）的常用选择。

同步：

import httpx


def fetch_json(url: str) -> dict:
    with httpx.Client(timeout=30) as client:   # with 自动关连接池
        response = client.get(url)
        response.raise_for_status()
        return response.json()

异步：

import httpx


async def fetch_json(url: str) -> dict:
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

async/await 是第 5 阶段（asyncio）的重点。这里先记住形：异步函数用 async def 定义，调用 IO 时 await，并用 async with 管理客户端。

4. 封装一个 API client（本阶段核心练习）

脚本里散落 requests.get(...) 很快会失控。正经做法是封一层 client，把 baseUrl、超时、重试、日志收在一处。这正是你在 Spring 里写 FooApiClient 的习惯。

参考实现见 scripts/stage3_http_client.py 里的 ApiClient，骨架：

class ApiClient:
    def __init__(self, base_url="", *, timeout=30, retry=None, client=None):
        self._retry = retry or RetryPolicy()
        # 注入点：测试传 MockTransport 的 client，生产传真 client
        self._client = client or httpx.Client(base_url=base_url, timeout=timeout)
        self._owns_client = client is None

    def __enter__(self): return self
    def __exit__(self, *exc): self.close()

    def get_json(self, url, *, params=None): ...
    def post_json(self, url, *, payload=None): ...

4.1 重试策略

@dataclass
class RetryPolicy:
    retries: int = 2          # 首次 + 最多再 2 次 = 共 3 次
    backoff_seconds: float = 0.2

重试的判断逻辑（关键认知）：

  • 5xx / 传输层错误（连不上、超时）→ 值得重试，可能是临时抖动。
  • 4xx 客户端错误（参数错、401、404）→ 不要重试，重试也还是错。
  • 每次重试前退避等待（backoff * attempt），避免把濒死的服务打得更惨。

4.2 依赖注入 = 可测试性

ApiClient.__init__ 接受一个 client= 参数。生产代码不传它，内部自建真 client；测试代码传一个挂了 MockTransport 的 client，整套重试逻辑都能离线验证。这和 Spring 构造注入 RestTemplate、测试换 mock 是同一思路。

5. 异步并发：一次打多个接口

批量调 embedding / 评估多个样本时，串行太慢。asyncio.gather 能并发跑一批协程：

import asyncio
import httpx


async def fetch_all(urls):
    async with httpx.AsyncClient(timeout=30) as client:
        responses = await asyncio.gather(
            *(client.get(url) for url in urls),
            return_exceptions=True,   # 单个失败不炸掉整批（错误隔离）
        )
    # responses 里可能混着 Response 和 Exception，逐个处理
    ...

return_exceptions=True 是并发版的"错误隔离"：某个 URL 失败时，它的异常被放进结果列表，而不是让整个 gather 抛出。参考实现 fetch_all 返回 [{url, ok, error, data}, ...]，和第 2 阶段评估脚本的结构化结果一脉相承。

Java 对照：类似 CompletableFuture.allOf(...) + 每个 future 单独 handle 异常，或 Reactor Flux.flatMap(...).onErrorResume(...)。

6. 测试 HTTP 代码（不打真网络）

真打网络的测试又慢又不稳定，绝不要这么做。两种离线手段：

6.1 httpx：MockTransport

import httpx


def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"pong": True})


client = httpx.Client(transport=httpx.MockTransport(handler))
# 把这个 client 注入 ApiClient，就能离线测全部逻辑（含重试）

handler 是个普通函数，按 request 返回你想要的 Response——想测重试就让它前两次返回 503、第三次返回 200。

6.2 requests：依赖注入 fake session

requests 没有内置 MockTransport，但我们的函数支持注入 session，测试里传一个 fake：

class _FakeSession:
    def get(self, url, params=None, timeout=None):
        return _FakeResponse({"id": 1})

完整测试见 tests/test_stage3_http_client.py。

7. 安全与习惯

  • API key 走环境变量，不要写死，日志里不要打印完整 key（见第 2 阶段）。
  • 每个请求都设 timeout。
  • 对外部依赖加重试 + 清晰的错误日志。
  • 复用 client / session，别每次请求都新建。
  • 解析逻辑（取字段、转换）和发请求分开写，解析部分能单独单测。

8. 可运行验证

参考脚本演示任务 1（调公开 API 并保存）：

.\.venv\Scripts\python.exe scripts\stage3_http_client.py --url https://jsonplaceholder.typicode.com/todos/1 --output data\stage3_todo.json

这条命令需要联网。任务 2/3/4 的逻辑都在离线测试里覆盖，无需网络。

运行测试（全部离线）：

.\.venv\Scripts\python.exe -m pytest tests\test_stage3_http_client.py

9. 本阶段掌握标准

  • 能用 requests 发同步 GET/POST，正确处理 timeout 和非 2xx。
  • 能用 httpx.Client 封装一个带重试和日志的 API client。
  • 能用 httpx.AsyncClient + asyncio.gather 并发请求并隔离单条错误。
  • 能解释 4xx 不重试、5xx 重试的原因。
  • 能用 httpx.MockTransport 或依赖注入给 HTTP 代码写离线测试。
  • 能读懂大多数 Python Agent 示例里的：模型 API 调用、embedding 调用、tool 调用、webhook、流式响应的基础结构。

10. 练习任务（自己动手）

参考脚本已实现 4 个任务，你可以在它基础上扩展：

  1. 给 ApiClient 加一个 headers 参数（如 Authorization: Bearer <key>，从环境变量读 key），并写测试验证 header 被发出。
  2. 给 fetch_all 加一个 concurrency 上限（用 asyncio.Semaphore），避免一次打 1000 个 URL 打爆对端。
  3. 给重试加"只对幂等方法（GET）重试，POST 默认不重试"的开关。
  4. 把响应里某个字段抽出来写成纯函数 parse_xxx(payload)，并单独单测。