深入理解Python httpx:现代HTTP客户端全解析_Python

python的httpx

httpx是一个功能强大且现代化的 python http 客户端库，支持同步和异步请求。它兼容 requestsapi，同时提供了更多高级特性（如 http/2、连接池、超时控制等）。以下是关于 httpx的核心知识点和使用指南：

一、安装 httpx

通过 pip 安装：

pip install httpx
# 如需支持 http/2，需额外安装依赖
pip install httpx[http2]

二、核心特性

同步与异步双模式：支持同步（httpx）和异步（asyncclient）请求。
http/2 支持：默认优先使用 http/2（需服务端支持）。
类型提示：完善的类型注解，提升代码可读性和 ide 支持。
连接池：复用 tcp 连接，减少握手开销。
超时控制：细粒度的超时设置（连接、读取、写入）。
代理与认证：支持 http/https/socks 代理、basic/digest/oauth 认证。
流式请求/响应：处理大文件上传/下载。
websocket 支持：实验性支持 websocket 协议。

三、基本用法

1. 同步请求（类似 requests）

使用 httpx.get()、post()等快捷方法，或创建 client实例管理连接。

示例：发送 get 请求

import httpx
# 快捷方式（适合简单请求）
response = httpx.get("https://httpbin.org/get")
print(response.status_code)  # 200
print(response.json())       # 解析 json 响应
# 使用 client 管理连接（推荐，复用连接更高效）
with httpx.client() as client:
    response = client.get("https://httpbin.org/get")
    print(response.text)

常用方法：

get(url, params=...)：get 请求
post(url, data=..., json=...)：post 请求（data传表单，json传 json）
put()、delete()、patch()：其他方法
head()、options()：head/options 请求

2. 异步请求（asyncclient）

需配合 async/await语法，适用于高并发场景。

示例：异步 get 请求

import asyncio
import httpx
async def fetch():
    async with httpx.asyncclient() as client:
        response = await client.get("https://httpbin.org/get")
        print(response.json())
# 运行异步函数
asyncio.run(fetch())

并发请求：利用 asyncio.gather实现批量并发：

async def main():
    urls = ["https://httpbin.org/get"] * 5
    async with httpx.asyncclient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        for resp in responses:
            print(resp.status_code)
asyncio.run(main())

四、高级配置

1. 超时设置

可全局设置或针对单个请求设置超时（单位：秒）：

# 单个请求超时（连接+读取总超时）
response = httpx.get("https://httpbin.org/delay/2", timeout=5.0)
# 细粒度超时（连接超时、读取超时、写入超时）
timeout = httpx.timeout(connect=5, read=10, write=2, pool=1)
response = httpx.get("https://httpbin.org/delay/2", timeout=timeout)
# client 级别超时
with httpx.client(timeout=timeout) as client:
    client.get("https://httpbin.org/delay/2")

2. 请求头与参数

请求头：通过 headers参数传递字典。
查询参数：通过 params参数传递字典（自动编码）。

headers = {"user-agent": "my-app/1.0"}
params = {"key1": "value1", "key2": "value2"}
response = httpx.get(
    "https://httpbin.org/get",
    headers=headers,
    params=params
)
# 实际请求 url：https://httpbin.org/get?key1=value1&key2=value2

3. post 请求（表单/json）

表单数据：使用 data参数（字典或字节流）。
json 数据：使用 json参数（自动序列化为 json 并设置 content-type: application/json）。

# 表单提交
data = {"username": "test", "password": "123"}
response = httpx.post("https://httpbin.org/post", data=data)
# json 提交
json_data = {"name": "alice", "age": 30}
response = httpx.post("https://httpbin.org/post", json=json_data)

4. 文件上传

使用 files参数传递文件对象（支持单文件或多文件）。

# 单文件上传
with open("file.txt", "rb") as f:
    files = {"file": ("file.txt", f, "text/plain")}  # (文件名, 文件对象, mime类型)
    response = httpx.post("https://httpbin.org/post", files=files)
# 多文件上传
files = [
    ("images", ("img1.jpg", open("img1.jpg", "rb"), "image/jpeg")),
    ("images", ("img2.png", open("img2.png", "rb"), "image/png"))
]
response = httpx.post("https://httpbin.org/post", files=files)

5. 代理设置

支持 http、https、socks 代理：

proxies = {
    "http://": "http://user:pass@proxy:8080",
    "https://": "https://user:pass@proxy:443",
    "all://": "socks5://user:pass@socks-proxy:1080"  # 所有协议走 socks5
}
# 全局代理
client = httpx.client(proxies=proxies)
# 单次请求代理
response = httpx.get("https://httpbin.org/ip", proxies=proxies)

6. 认证（authentication）

支持 basic、digest、oauth 等认证方式：

from httpx import basicauth
# basic 认证
auth = basicauth(username="user", password="pass")
response = httpx.get("https://httpbin.org/basic-auth/user/pass", auth=auth)
# 或直接传递元组（自动识别为 basicauth）
response = httpx.get("https://httpbin.org/basic-auth/user/pass", auth=("user", "pass"))

7. 流式响应（streaming）

处理大文件下载时，避免一次性加载到内存：

with httpx.stream("get", "https://httpbin.org/stream/20") as response:
    for chunk in response.iter_bytes(chunk_size=1024):  # 按块读取
        print(len(chunk))  # 处理每个块（如写入文件）

五、异常处理

httpx定义了丰富的异常类，需捕获特定异常而非通用 exception：

异常类	描述
`httpx.requesterror`	网络层错误（如 dns 失败、连接拒绝）
`httpx.httpstatuserror`	响应状态码非 2xx（需手动触发）
`httpx.timeoutexception`	超时错误
`httpx.connecterror`	连接错误
`httpx.readerror`	读取响应错误

示例：异常处理

try:
    response = httpx.get("https://httpbin.org/status/404")
    response.raise_for_status()  # 若状态码非 2xx，抛出 httpstatuserror
except httpx.httpstatuserror as e:
    print(f"http 错误: {e.response.status_code}")
except httpx.requesterror as e:
    print(f"请求错误: {e}")
except httpx.timeoutexception:
    print("请求超时")

六、与 requests 对比

特性	requests	httpx
异步支持	❌ 不支持	✅ 原生支持（asyncclient）
http/2 支持	❌ 不支持	✅ 支持（需安装依赖）
连接池	基础支持	✅ 更高效的连接池管理
类型提示	有限	✅ 完善
websocket	❌ 不支持	✅ 实验性支持