> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# httpx 完全指南
- URL: https://blog.vercanti.com/httpx-wan-quan-zhi-nan/
- Published: 2026-08-28T14:34:47.000Z
- Updated: 2026-08-28T14:57:19.000Z
- Description: httpx 是现代 Python HTTP 客户端，同时支持同步和异步，是 requests 库的异步升级替代品。 使用 Client 可以复用连接、共享配置（headers、cookies、超时等）： httpx 本身不内置重试，推荐配合 tenacity： 每个 Client 维护自己的连接池，频繁创建/销毁会导致连接无法复用，性能差。应在应用启动时创建，贯穿整个生命周期（或通过 FastAPI 依赖注入共享）。 生产代码使用 AsyncClient 作为应用级单例：AsyncClient 内部维护连接池，每次请求新建实例会频繁建立 TCP 连接，应
- Author: yellowdog
- Tags: Python, 框架与库

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

---

## 1\. 基础概念

### httpx 是什么

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

| 特性     | requests | httpx                  |
| ------ | -------- | ---------------------- |
| 异步支持   | 无        | 原生支持（AsyncClient）      |
| HTTP/2 | 无        | 支持（需安装 httpx\[http2\]） |
| 类型注解   | 较少       | 完整                     |
| 连接池    | 有        | 有（Client 级别）           |
| 超时控制   | 简单       | 细粒度（连接/读/写/池）          |

### 安装

```bash
pip install httpx

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

```

---

## 2\. 基础使用

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

```python
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、超时等）：

```python
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\. 请求参数

### 请求方法与参数

```python
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.status\_code         | HTTP 状态码                    |
| r.headers              | 响应头（字典）                     |
| r.cookies              | 响应 Cookies                  |
| r.text                 | 响应文本（自动检测编码）                |
| r.content              | 响应二进制内容                     |
| r.json()               | 解析 JSON，返回字典                |
| r.url                  | 最终请求 URL（含重定向）              |
| r.encoding             | 检测到的编码                      |
| r.raise\_for\_status() | 4xx/5xx 时抛出 HTTPStatusError |
| r.is\_success          | 状态码 2xx                     |
| r.is\_redirect         | 状态码 3xx                     |
| r.elapsed              | 请求耗时（timedelta）             |

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

```

---

## 4\. Client 配置

### Client 初始化参数

| 参数                | 类型              | 说明                         |
| ----------------- | --------------- | -------------------------- |
| base\_url         | str             | 基础 URL，所有请求路径自动拼接          |
| headers           | dict            | 默认请求头（每次请求都会携带）            |
| cookies           | dict            | 默认 Cookies                 |
| timeout           | float / Timeout | 超时设置                       |
| follow\_redirects | bool            | 是否自动跟随重定向，默认 False         |
| verify            | bool / str      | SSL 验证，False 关闭，str 指定证书路径 |
| proxy             | str             | 单一代理 URL，所有流量通过该代理路由       |
| mounts            | dict            | 自定义传输挂载，可按 URL 前缀分别设置代理    |
| auth              | tuple / Auth    | 认证（Basic Auth 等）           |
| limits            | Limits          | 连接池限制                      |
| http2             | bool            | 启用 HTTP/2                  |
| event\_hooks      | dict            | 请求/响应钩子                    |

### 超时配置

```python
# 统一超时（秒）
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)

```

### 连接池限制

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

```

---

## 5\. 认证

### 内置认证

```python
# 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"))

```

### 自定义认证类

```python
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）

```python
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`：

```bash
pip install tenacity

```

```python
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\. 流式传输

### 流式响应（大文件下载）

```python
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）

```python
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\. 代理设置

```python
# 全局代理（单一代理，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 客户端

```python
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()

```

### 并发请求限速

```python
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\. 最佳实践

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

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

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

```

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

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

### raise\_for\_status() 放在响应后立即调用

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

```

---

## 12\. 踩坑与注意事项

### 不带 base\_url 时，路径不要以 / 开头

```python
# 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 验证关闭仅限测试环境

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

```

---

## 最佳实践

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

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

```

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

```python
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` 上下文管理器**：避免将整个响应体加载到内存：

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

```

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

```python
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 相同）：

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

```

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

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

---

## 参见

- [asyncio异步编程完全指南](https://blog.vercanti.com/asyncio-yi-bu-bian-cheng-wan-quan-zhi-nan/)
- [FastAPI完全指南](https://blog.vercanti.com/fastapi-wan-quan-zhi-nan/)
- [Scrapy分布式采集](https://blog.vercanti.com/scrapy-fen-bu-shi-cai-ji-wan-quan-zhi-nan/)
- [Playwright完全指南](https://blog.vercanti.com/playwright-wan-quan-zhi-nan/)