httpx 完全指南

官方文档：https://www.python-httpx.org/
最后更新：2026-03-29

1. 基础概念

httpx 是什么

httpx 是现代 Python HTTP 客户端，同时支持同步和异步，是 requests 库的异步升级替代品。

安装

pip install httpx

# 可选功能
pip install httpx[http2]    # HTTP/2 支持
pip install httpx[brotli]   # Brotli 压缩支持

2. 基础使用

单次请求（不推荐用于生产）

import httpx

# 同步
response = httpx.get("https://httpbin.org/get")
response = httpx.post("https://httpbin.org/post", json={"key": "value"})

# 异步
import asyncio

async def main():
    response = await httpx.get("https://httpbin.org/get")  # 不推荐，每次创建新连接
asyncio.run(main())

Client / AsyncClient（推荐）

使用 Client 可以复用连接、共享配置（headers、cookies、超时等）：

import httpx

# 同步客户端（上下文管理器自动关闭连接）
with httpx.Client(base_url="https://api.example.com") as client:
    response = client.get("/users")
    response = client.post("/users", json={"name": "Alice"})

# 异步客户端
import asyncio

async def main():
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        response = await client.get("/users")
        data = response.json()

asyncio.run(main())

3. 请求参数

请求方法与参数

async with httpx.AsyncClient() as client:
    # GET 查询参数
    r = await client.get("/search", params={"q": "python", "page": 1})
    # 实际 URL: /search?q=python&page=1

    # POST JSON 请求体
    r = await client.post("/users", json={"name": "Alice", "age": 25})

    # POST 表单
    r = await client.post("/login", data={"username": "alice", "password": "secret"})

    # POST 文件上传
    with open("photo.jpg", "rb") as f:
        r = await client.post("/upload", files={"photo": f})

    # 多文件 + 表单混合
    r = await client.post("/upload", files={
        "photo": ("photo.jpg", open("photo.jpg", "rb"), "image/jpeg"),
    }, data={"description": "头像"})

    # 自定义请求头
    r = await client.get("/protected", headers={"Authorization": "Bearer token123"})

    # 发送 cookies
    r = await client.get("/profile", cookies={"session": "abc123"})

Response 对象

r = await client.get("/users")
r.raise_for_status()  # 非 2xx 时抛出异常
users = r.json()

4. Client 配置

Client 初始化参数

超时配置

# 统一超时（秒）
client = httpx.AsyncClient(timeout=10.0)

# 细粒度超时
timeout = httpx.Timeout(
    connect=5.0,   # 建立连接的超时
    read=30.0,     # 读取响应的超时
    write=10.0,    # 发送请求的超时
    pool=5.0,      # 等待连接池空闲的超时
)
client = httpx.AsyncClient(timeout=timeout)

# 单次请求覆盖
r = await client.get("/slow", timeout=60.0)

# 禁用超时（不推荐）
r = await client.get("/stream", timeout=None)

连接池限制

limits = httpx.Limits(
    max_connections=100,      # 最大连接数
    max_keepalive_connections=20,  # 最大保活连接数
    keepalive_expiry=30,      # 保活连接过期时间（秒）
)
client = httpx.AsyncClient(limits=limits)

5. 认证

内置认证

# Basic Auth
client = httpx.AsyncClient(auth=("username", "password"))
r = await client.get("/protected")

# Bearer Token（两种写法）
client = httpx.AsyncClient(headers={"Authorization": "Bearer my-token"})

r = await client.get("/protected", auth=("user", "pass"))

自定义认证类

import httpx

class BearerAuth(httpx.Auth):
    def __init__(self, token: str):
        self.token = token

    def auth_flow(self, request: httpx.Request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        yield request

class JWTRefreshAuth(httpx.Auth):
    """Token 过期时自动刷新的认证"""
    def __init__(self, token: str, refresh_token: str):
        self.token = token
        self.refresh_token = refresh_token

    def auth_flow(self, request: httpx.Request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        response = yield request  # 发送请求，获取响应

        if response.status_code == 401:
            # 刷新 token
            refresh_response = yield self._build_refresh_request()
            self.token = refresh_response.json()["access_token"]
            # 用新 token 重试原始请求
            request.headers["Authorization"] = f"Bearer {self.token}"
            yield request

    def _build_refresh_request(self):
        return httpx.Request(
            "POST",
            "https://api.example.com/auth/refresh",
            json={"refresh_token": self.refresh_token},
        )

client = httpx.AsyncClient(auth=BearerAuth("my-token"))

6. 事件钩子（Hook）

import httpx

async def log_request(request: httpx.Request):
    print(f">>> {request.method} {request.url}")

async def log_response(response: httpx.Response):
    await response.aread()  # 确保响应体已读取
    print(f"<<< {response.status_code} {response.url} ({response.elapsed.total_seconds():.3f}s)")

client = httpx.AsyncClient(
    event_hooks={
        "request": [log_request],
        "response": [log_response],
    }
)

7. 重试机制（配合 tenacity）

httpx 本身不内置重试，推荐配合 tenacity：

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)),
    reraise=True,
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> dict:
    r = await client.get(url, timeout=10)
    r.raise_for_status()
    return r.json()

8. 流式传输

流式响应（大文件下载）

async with httpx.AsyncClient() as client:
    async with client.stream("GET", "https://example.com/large-file.zip") as r:
        r.raise_for_status()
        with open("file.zip", "wb") as f:
            async for chunk in r.aiter_bytes(chunk_size=8192):
                f.write(chunk)

流式 JSON（如 SSE / NDJSON）

async with client.stream("GET", "/stream") as r:
    async for line in r.aiter_lines():
        if line.startswith("data:"):
            data = json.loads(line[5:])
            print(data)

9. 代理设置

# 全局代理（单一代理，0.28+ 新 API）
client = httpx.AsyncClient(proxy="http://127.0.0.1:7890")

# 按协议分别设置（使用 mounts）
client = httpx.AsyncClient(mounts={
    "http://": httpx.AsyncHTTPTransport(proxy="http://proxy:7890"),
    "https://": httpx.AsyncHTTPTransport(proxy="http://proxy:7890"),
})

# 忽略某些域名的代理（不走代理）
client = httpx.AsyncClient(mounts={
    "https://internal.company.com": None,
    "all://": httpx.AsyncHTTPTransport(proxy="http://proxy:7890"),
})

10. 常用代码段

封装通用 API 客户端

import httpx
from typing import Any

class APIClient:
    def __init__(self, base_url: str, token: str):
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=httpx.Timeout(connect=5.0, read=30.0),
        )

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self._client.aclose()

    async def get(self, path: str, **kwargs) -> Any:
        r = await self._client.get(path, **kwargs)
        r.raise_for_status()
        return r.json()

    async def post(self, path: str, **kwargs) -> Any:
        r = await self._client.post(path, **kwargs)
        r.raise_for_status()
        return r.json()

并发请求限速

import asyncio
import httpx

async def fetch_all(urls: list[str], concurrency: int = 10) -> list[dict]:
    sem = asyncio.Semaphore(concurrency)

    async def fetch(client: httpx.AsyncClient, url: str) -> dict:
        async with sem:
            r = await client.get(url, timeout=10)
            r.raise_for_status()
            return r.json()

    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

11. 最佳实践

始终使用上下文管理器或手动关闭

# 正确：with 语句自动关闭
async with httpx.AsyncClient() as client:
    ...

# 或手动关闭
client = httpx.AsyncClient()
try:
    ...
finally:
    await client.aclose()

不要为每次请求创建新的 Client

每个 Client 维护自己的连接池，频繁创建/销毁会导致连接无法复用，性能差。应在应用启动时创建，贯穿整个生命周期（或通过 FastAPI 依赖注入共享）。

raise_for_status() 放在响应后立即调用

r = await client.get(url)
r.raise_for_status()  # 早失败，避免用 None 或错误数据继续执行
data = r.json()

12. 踩坑与注意事项

不带 base_url 时，路径不要以 / 开头

# base_url="https://api.example.com/v1"
client.get("/users")   # 结果：https://users（错误！路径替换了整个 base）
client.get("users")    # 结果：https://api.example.com/v1/users（正确）

# 正确做法：base_url 以 / 结尾，路径不以 / 开头
client = httpx.AsyncClient(base_url="https://api.example.com/v1/")
client.get("users")    # https://api.example.com/v1/users

SSL 验证关闭仅限测试环境

# 开发/测试时忽略 SSL 证书验证
client = httpx.AsyncClient(verify=False)
# 生产环境必须开启（默认开启），或指定自定义 CA 证书
client = httpx.AsyncClient(verify="/path/to/ca-bundle.crt")

最佳实践

生产代码使用 AsyncClient 作为应用级单例：AsyncClient 内部维护连接池，每次请求新建实例会频繁建立 TCP 连接，应在应用启动时创建一个实例并复用：

# FastAPI 集成示例
@asynccontextmanager
async def lifespan(app):
    app.state.http = httpx.AsyncClient(timeout=30.0)
    yield
    await app.state.http.aclose()

始终设置 timeout 参数：httpx 默认超时 5 秒，生产环境按业务 SLA 设置，区分连接超时和读取超时：

timeout = httpx.Timeout(connect=3.0, read=30.0, write=10.0, pool=5.0)
client = httpx.AsyncClient(timeout=timeout)

用 client.build_request() + client.send() 实现请求重试：分两步构造和发送请求，方便在重试逻辑中复用同一 Request 对象，而不是重新构造。

流式下载大文件用 stream 上下文管理器：避免将整个响应体加载到内存：

async with client.stream("GET", url) as resp:
    async for chunk in resp.aiter_bytes(chunk_size=8192):
        await file.write(chunk)

用 httpx.MockTransport 做单元测试：不依赖真实网络，替换传输层实现确定性测试：

transport = httpx.MockTransport(handler=lambda req: httpx.Response(200, json={"ok": True}))
client = httpx.AsyncClient(transport=transport)

常见陷阱

陷阱：AsyncClient 未关闭导致资源泄漏

现象： 程序退出时报 Unclosed client，或长时间运行后文件描述符耗尽。
原因： AsyncClient 持有连接池，需要显式关闭。直接实例化而不用 async with 或不调用 aclose() 时泄漏。
解决： 总是用 async with httpx.AsyncClient() as client，或在应用关闭时调用 await client.aclose()。

陷阱：在同步函数中调用 AsyncClient

现象： 在非 async 函数中使用 AsyncClient 报 RuntimeError: This event loop is already running 或 coroutine 未等待。
原因： AsyncClient 的方法是协程，必须在 async 上下文中 await。
解决： 同步场景改用 httpx.Client（同步版本，API 相同）：

with httpx.Client(timeout=10.0) as client:
    resp = client.get(url)

陷阱：重定向跟随导致意外发送认证头到第三方

现象： 携带 Authorization 头的请求被 301 重定向到其他域后，认证信息被发送到新域名。
原因： httpx 默认跟随重定向且不清除认证头（与 requests 不同）。
解决： 敏感请求设置 follow_redirects=False，或使用 trust_env=False 配合自定义 auth 参数精确控制。

参见

• asyncio异步编程完全指南
• FastAPI完全指南
• Scrapy分布式采集
• Playwright完全指南