Python22_httpx网络请求

httpx.AsyncClient，这是 Python 中处理异步 HTTP 请求的核心类。

httpx.AsyncClient 概览

httpx.AsyncClient 是 HTTPX 库的异步客户端，基于 asyncio 构建，支持非阻塞的 HTTP/1.1 和 HTTP/2 请求，性能远超同步的 requests 库。

---

基本使用

1. 安装

pip install httpx
# 如需 HTTP/2 支持
pip install httpx[http2]

2. 简单请求

import httpx
import asyncio

async def main():
    # 创建客户端（推荐用 async with 管理生命周期）
    async with httpx.AsyncClient() as client:
        # GET 请求
        response = await client.get("https://api.github.com")
        print(response.status_code)  # 200
        print(response.json())       # 解析 JSON

asyncio.run(main())

---

核心特性详解

3. 请求方法

async with httpx.AsyncClient() as client:
    # 各类 HTTP 方法
    r1 = await client.get(url, params={"key": "value"})
    r2 = await client.post(url, json={"data": "value"})      # JSON 体
    r3 = await client.post(url, data={"form": "field"})      # 表单数据
    r4 = await client.put(url, content=b"raw bytes")         # 原始字节
    r5 = await client.patch(url, files={"file": open("a.jpg", "rb")})
    r6 = await client.delete(url)
    r7 = await client.head(url)
    r8 = await client.options(url)

4. 高级配置

client = httpx.AsyncClient(
    # 超时设置（连接、读取、写入）
    timeout=httpx.Timeout(10.0, connect=5.0),
    
    # 请求头
    headers={"User-Agent": "MyApp/1.0"},
    
    # 基础 URL（后续请求可写相对路径）
    base_url="https://api.example.com/v1",
    
    # 启用 HTTP/2
    http2=True,
    
    # 跟随重定向
    follow_redirects=True,
    
    # 验证 SSL（开发时可关闭）
    verify=True,  # 或 verify=False（不推荐生产环境）
    
    # 代理
    proxies="http://localhost:8080",
    # 或按协议区分：proxies={"http://": "...", "https://": "..."}
    
    # 认证
    auth=("username", "password"),  # Basic Auth
    # 或 auth=httpx.BearerToken("token")
    
    # Cookie 持久化
    cookies={"session": "abc123"},
    
    # 限制连接池
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)

# 使用 base_url 后
response = await client.get("/users")  # 实际请求 https://api.example.com/v1/users

5. 并发请求（核心优势）

import asyncio
import httpx

async def fetch(client, url):
    response = await client.get(url)
    return response.json()

async def main():
    urls = [
        "https://api.github.com/users/octocat",
        "https://api.github.com/users/torvalds",
        "https://api.github.com/users/gvanrossum"
    ]
    
    async with httpx.AsyncClient() as client:
        # 并发执行所有请求
        tasks = [fetch(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(results)

asyncio.run(main())

6. 流式响应（大文件下载）

async with httpx.AsyncClient() as client:
    # 流式读取，避免内存溢出
    async with client.stream("GET", "https://example.com/large-file.zip") as response:
        async for chunk in response.aiter_bytes():
            # 处理每个数据块
            print(f"Received {len(chunk)} bytes")
            
    # 或流式文本
    async with client.stream("GET", url) as response:
        async for line in response.aiter_lines():
            print(line)

7. 请求/响应钩子

def log_request(request):
    print(f"→ Request: {request.method} {request.url}")

def log_response(response):
    print(f"← Response: {response.status_code}")

async with httpx.AsyncClient(
    event_hooks={
        "request": [log_request],
        "response": [log_response]
    }
) as client:
    await client.get("https://httpbin.org/get")

---

与同步 Client 对比

特性	httpx.Client	httpx.AsyncClient
执行方式	同步阻塞	异步非阻塞
适用场景	脚本、简单任务	高并发、Web 服务
性能	一般	极高（可处理数千并发）
使用方式	with client:	async with client:
依赖	无额外依赖	需要 asyncio

---

最佳实践

import httpx
import asyncio

class APIClient:
    """封装示例：带重试和错误处理的异步客户端"""
    
    def __init__(self):
        self.client = httpx.AsyncClient(
            base_url="https://api.example.com",
            timeout=30.0,
            http2=True,
            limits=httpx.Limits(max_connections=50)
        )
    
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, *args):
        await self.client.aclose()  # 确保关闭
    
    async def get_user(self, user_id: int):
        try:
            response = await self.client.get(f"/users/{user_id}")
            response.raise_for_status()  # 自动抛出 4xx/5xx 异常
            return response.json()
        except httpx.HTTPStatusError as e:
            print(f"HTTP error: {e.response.status_code}")
            raise
        except httpx.RequestError as e:
            print(f"Request failed: {e}")
            raise

# 使用
async def main():
    async with APIClient() as api:
        user = await api.get_user(123)
        print(user)

asyncio.run(main())

---

常见异常处理

异常	说明
httpx.RequestError	网络连接错误
httpx.HTTPStatusError	HTTP 错误状态码（需调用 raise_for_status()）
httpx.TimeoutException	请求超时
httpx.ConnectError	连接失败

httpx.AsyncClient 是现代 Python 异步编程中处理 HTTP 请求的首选工具，特别适合 FastAPI/Starlette 等异步 Web 框架的后端服务调用。

---