社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
aigc
aigc   chatgpt  
WEB开发
linux   MongoDB   Redis   DATABASE   NGINX   其他Web框架   web工具   zookeeper   tornado   NoSql   Bootstrap   js   peewee   Git   bottle   IE   MQ   Jquery  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广  
产品
短视频  
印度
印度  
Py学习  »  Python

Httpx, 一个神奇的 Python 库-Client

程序员老朱 • 10 月前 • 541 次点击  

如果你从 Requests 库迁移而来,httpx.Client() 相当于 requests.Session(),是管理 HTTP 会话的核心类。

除了简单的实验、一次性脚本或原型开发外,「其它操作我们都应该使用 Client 实例」。一次性脚本的顶层API示例

「更高效的网络资源利用」

使用顶层 API(如 httpx.get())时,HTTPX 需要为每个请求建立新的连接(连接不会被复用)。随着向同一主机发起的请求增多,这种方式会迅速变得低效。

相比之下,Client 实例使用 HTTP 连接池。这意味着当你向同一主机发起多个请求时,Client 会重用底层的 TCP 连接,而不是为每个请求重新创建连接。

与使用顶层 API 相比,这能带来显著的性能提升:

  1. 「降低请求延迟」(无需重复握手)
  2. 「减少 CPU 使用和往返次数」
  3. 「减轻网络拥塞」

「额外功能」

Client 实例还支持顶层 API 不具备的功能:

  1. 跨请求的 Cookie 持久化
  2. 为所有请求应用统一配置
  3. 通过 HTTP 代理发送请求
  4. 使用 HTTP/2 协议

基本用法

讲解:使用上下文管理器

使用上下文管理器是推荐的方式,它能确保连接在离开  with 块时被正确清理,避免资源泄漏。

import httpx

with httpx.Client() as client:
    response = client.get('http://httpbin.org/get')
    print(response.status_code)

讲解:显式关闭方式

如果你需要在多个函数或作用域中使用客户端,可以显式管理其生命周期,但需要确保正确关闭。

client = httpx.Client()
try:
    response = client.get('http://httpbin.org/get')
    print(response.json())
finally:
    client.close()  # 确保连接池被关闭

发送请求

讲解:基本请求方法

Client 实例提供与顶层 API 相同的方法(.get()、.post() 等),接受相同参数。

with httpx.Client() as client:
    # GET 请求
    response = client.get('http://httpbin.org/get')
    print(f"GET 响应: {response.status_code}")
    
    # POST 请求
    data = {'name': 'Alice', 'email': 'alice@example.com'}
    response = client.post('http://httpbin.org/post', json=data)
    print(f"POST 响应: {response.status_code}")
    
    # 带参数的请求
    params = {'page': 1, 'limit': 20}
    response = client.get('http://httpbin.org/get', params=params)
    print(f"带参数的请求: {response.url}")

讲解:自定义请求头

你可以为每个请求设置特定的请求头,这些头信息将包含在 HTTP 请求中。




    
with httpx.Client() as client:
    headers = {
        'X-Custom-Header': 'my-value',
        'User-Agent': 'MyApp/1.0',
        'Accept': 'application/json'
    }
    response = client.get('http://httpbin.org/headers', headers=headers)
    print("返回的请求头:", response.json()['headers'])

跨请求共享配置

讲解:全局请求头配置

通过在 Client 构造函数中传递参数,可以将配置应用到所有出站请求,简化代码并确保一致性。

# 为所有请求设置公共请求头
headers = {
    'User-Agent': 'my-app/0.0.1',
    'Accept-Language': 'zh-CN,zh;q=0.9',
    'X-Client-ID': 'CLIENT123'
}

with httpx.Client(headers=headers, timeout=30.0) as client:
    # 这个请求会自动包含上面的请求头
    r1 = client.get('http://httpbin.org/headers')
    print("第一个请求的请求头:", r1.json()['headers'])
    
    # 这个请求也会包含
    r2 = client.get('http://httpbin.org/user-agent')
    print("User-Agent:", r2.json()['user-agent'])

讲解:基础 URL 配置

base_url 参数允许为所有出站请求添加 URL 前缀,简化代码并提高可维护性。

# 设置基础 URL,简化后续请求
with httpx.Client(base_url='http://httpbin.org') as client:
    # 实际请求: http://httpbin.org/get
    response1 = client.get('/get')
    print(f"响应1: {response1.status_code}")
    
    # 实际请求: http://httpbin.org/post
    response2 = client.post('/post', json={'test': 'data'})
    print(f"响应2:  {response2.status_code}")
    
    # 仍然可以请求完整 URL(不会添加基础 URL)
    external = client.get('https://example.com')
    print(f"外部请求: {external.status_code}")

配置合并规则

讲解:参数合并方式

当客户端级别和请求级别都有配置时,了解合并规则很重要。头部、查询参数和 Cookie 会合并,其他参数以请求级别优先。

# 客户端级别配置
with httpx.Client(
    headers={'X-Auth': 'client-token'},
    params={'api_version': 'v2'}
) as client:
    
    # 请求级别配置(会与客户端配置合并)
    response = client.get(
        'http://httpbin.org/anything',
        headers={'X-Custom': 'request-value'},
        params={'page': 1}
    )
    
    result = response.json()
    print("合并后的查询参数:", result['args'])
    print("合并后的请求头:", result['headers'])

讲解:参数覆盖规则

除头部、查询参数和 Cookie 外,其他参数在请求级别会覆盖客户端级别配置。

# 认证信息:请求级别覆盖客户端级别
with httpx.Client(auth=('cuser','cpass')) as client:
    # 使用请求级别的认证,而非客户端级别的
    response = client.get(
        'http://httpbin.org/basic-auth/user/pass',
        auth=('user', 'pass')
    )
    
    if response.status_code == 200:
        print("认证成功:", response.json())
    else:
        print("认证失败")

客户端独有配置选项

讲解:HTTP/2 支持

HTTPX 支持 HTTP/2 协议,可以通过简单配置启用,提升性能(特别是在高并发场景)。




    
# 启用 HTTP/2(如果服务器支持)
with httpx.Client(http2=True) as client:
    response = client.get('https://httpbin.org/get')
    print(f"HTTP 版本: {response.http_version}")

讲解:代理配置

通过代理发送请求是常见需求,HTTPX 提供了简单的配置方式。

# 通过代理发送请求
client_config = {
    "timeout": 30.0,
    "proxy": "http://localhost:7888",
}

with httpx.Client(**client_config) as client:
    response = client.get('http://httpbin.org/get')
    print(f"通过代理的请求状态: {response.status_code}")

讲解:连接限制配置

对于高并发应用,合理配置连接池限制可以优化性能和资源使用。

# 限制连接数和并发数
limits = httpx.Limits(
    max_connections=100,      # 最大连接数
    max_keepalive_connections=20,  # 最大保持活动的连接数
    keepalive_expiry=5.0      # 连接保持时间(秒)
)

with httpx.Client(limits=limits) as client:
    # 适用于高并发场景
    for i in range(6):
       response = client.get(f'http://httpbin.org/delay/{i%3}')
       print(f"请求 {i+1}: {response.status_code}")

使用 Request 实例

讲解:构建 Request 对象

为了最大程度控制网络传输内容,HTTPX 支持构建显式的 Request 实例,提供更细粒度的控制。




    
# 创建请求对象
request = httpx.Request(
    method="POST",
    url="http://httpbin.org/post",
    json={"key": "value"},
    headers={"X-Custom": "header-value"}
)

# 发送请求
with httpx.Client() as client:
    response = client.send(request)
    print(f"请求方法: {response.request.method}")
    print(f"响应状态: {response.status_code}")

讲解:高级请求控制

通过构建请求对象并修改它,可以实现更复杂的请求逻辑,如动态修改头信息。

# 初始配置
default_headers = {
    "X-API-Key": "default-key",
    "X-Client-ID": "CLIENT123"
}

with httpx.Client(headers=default_headers) as client:
    # 构建基础请求
    request = client.build_request("GET", "http://httpbin.org/headers")
    
    print("原始请求头中的 X-Client-ID:", request.headers.get("X-Client-ID"))
    
    # 为特定请求移除敏感头信息
    del request.headers["X-API-Key"]
    
    # 添加请求特定头
    request.headers["X-Request-ID"] = "req-12345"
    
    # 发送修改后的请求
    response = client.send(request)
    headers = response.json()['headers']
    print("发送的请求头:", {k: v for k, v in headers.items() if k.startswith('X-')})

监控下载进度

讲解:下载进度监控

对于大文件下载,可以使用响应流式传输并检查进度,tqdm 库提供了美观的进度条。

import httpx
from tqdm import tqdm

# 下载文件并显示进度
url = "http://httpbin.org/stream-bytes/1000000"# 1MB 数据

with httpx.Client() as client:
    with client.stream("GET", url) as response:
        # 获取文件总大小
        total = int(response.headers.get("Content-Length", 0))

        # 创建进度条
        with tqdm(total=total, unit="B", unit_scale=True, desc="下载") as pbar:
            for chunk in response.iter_bytes():
                # 在这里处理数据,比如保存到文件
                # file.write(chunk)

                # 更新进度条
                pbar.update(len(chunk))

        print("下载完成!")

讲解:上传进度监控

同样,你可以监控大文件的上传进度,使用生成器函数逐步产生数据。

import httpx
import io
from tqdm import tqdm

# 生成测试数据
data = b"x" * 1000000# 1MB 数据


def generate_data():
    """生成器函数,用于上传数据"""
    buffer = io.BytesIO(data)
    chunk_size = 8192

    # 创建进度条
    with tqdm(total=len(data), unit="B", unit_scale=True, desc="上传") as pbar:
        whileTrue:
            chunk = buffer.read(chunk_size)
            ifnot chunk:
                break
            yield chunk
            pbar.update(len(chunk))


# 发送上传请求
with httpx.Client() as client:
    response = client.post("http://httpbin.org/post", content=generate_data())
    print(f"上传完成,响应状态: {response.status_code}")

总结

HTTPX 的 Client 类提供了强大而灵活的功能,适用于生产环境中的各种 HTTP 通信需求。通过合理使用连接池、配置合并、进度监控和流式处理等特性,可以构建高效、可靠且易于维护的网络应用程序。

记住关键原则:

  • 「始终使用上下文管理器」确保资源正确释放
  • 「根据场景合理配置超时和连接池」
  • 「大文件使用流式传输」避免内存问题
  • 「利用 Client 的配置共享」简化代码
  • 「监控进度和统计信息」优化用户体验和性能

使用 httpbin.org 作为测试端点可以帮助你理解和验证 HTTP 请求的各个方面,是学习和调试 HTTP 客户端的绝佳工具。

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/190867