---

name: httpx description: 面向Python的下一代HTTP客户端，同时支持同步和异步，完美适用于现代Python应用程序 when_to_use: 当您需要在Python中进行HTTP请求时使用，特别是涉及异步/等待、流式传输或高级身份验证的场景

HTTPX 技能

HTTPX 是一个功能齐全的Python HTTP客户端，提供同步和异步API，并支持HTTP/1.1和HTTP/2。

快速开始

基本用法

import httpx

# 简单的GET请求
response = httpx.get('https://api.example.com/data')
print(response.status_code)
print(response.json())

# 发送JSON数据的POST请求
response = httpx.post('https://api.example.com/users', json={'name': 'Alice'})

异步用法

import asyncio
import httpx

async def fetch_data():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        return response.json()

result = asyncio.run(fetch_data())

常见模式

1. 带连接池的异步客户端

import httpx

async def make_multiple_requests():
    async with httpx.AsyncClient() as client:
        # 复用同一个客户端进行多次请求
        tasks = [
            client.get('https://api.example.com/users/1'),
            client.get('https://api.example.com/users/2'),
            client.get('https://api.example.com/users/3')
        ]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

2. 身份验证

import httpx

# 基本身份验证
auth = httpx.BasicAuth(username='user', password='pass')
client = httpx.Client(auth=auth)

# Bearer令牌身份验证
headers = {'Authorization': 'Bearer your-token-here'}
client = httpx.Client(headers=headers)

# 按请求身份验证
response = client.get('https://api.example.com', auth=('user', 'pass'))

3. 流式下载

import httpx

# 流式传输大文件而不加载到内存中
with httpx.stream('GET', 'https://example.com/large-file.zip') as response:
    with open('large-file.zip', 'wb') as f:
        for chunk in response.iter_bytes():
            f.write(chunk)

# 异步流式传输
async def download_large_file():
    async with httpx.AsyncClient() as client:
        async with client.stream('GET', 'https://example.com/large-file.zip') as response:
            with open('large-file.zip', 'wb') as f:
                async for chunk in response.aiter_bytes():
                    f.write(chunk)

4. 流式上传

import httpx

# 流式上传大文件
async def upload_large_file():
    def generate_data():
        # 生成数据块
        for i in range(100):
            yield f'chunk {i}
'.encode()

    async with httpx.AsyncClient() as client:
        response = await client.post(
            'https://api.example.com/upload',
            content=generate_data()
        )
        return response

5. 错误处理和超时

import httpx

# 配置超时
client = httpx.Client(
    timeout=httpx.Timeout(10.0, connect=5.0, read=8.0)
)

try:
    response = client.get('https://api.example.com/slow')
    response.raise_for_status()  # 对于4XX/5XX响应抛出异常
except httpx.TimeoutException:
    print("请求超时")
except httpx.HTTPStatusError as e:
    print(f"HTTP错误: {e.response.status_code}")
except httpx.RequestError as e:
    print(f"请求失败: {e}")

6. 客户端配置

import httpx

# 带共享配置的客户端
client = httpx.Client(
    base_url='https://api.example.com',
    headers={'User-Agent': 'MyApp/1.0'},
    timeout=30.0,
    follow_redirects=True
)

# 所有请求都将使用base_url和headers
response = client.get('/users')  # 向 https://api.example.com/users 发送请求

7. 自定义身份验证

import httpx

class CustomAuth(httpx.Auth):
    def __init__(self, api_key):
        self.api_key = api_key

    def auth_flow(self, request):
        request.headers['X-API-Key'] = self.api_key
        yield request

# 使用自定义身份验证
auth = CustomAuth('your-secret-api-key')
client = httpx.Client(auth=auth)

8. 进度监控

import httpx
from tqdm import tqdm

def download_with_progress(url, filename):
    with httpx.stream('GET', url) as response:
        total = int(response.headers.get('content-length', 0))

        with tqdm(total=total, unit='B', unit_scale=True) as progress:
            with open(filename, 'wb') as f:
                for chunk in response.iter_bytes():
                    f.write(chunk)
                    progress.update(len(chunk))

9. 重试逻辑

import httpx
import time

def make_request_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = httpx.get(url, timeout=10.0)
            response.raise_for_status()
            return response
        except httpx.RequestError as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避

10. WebSocket支持（使用httpx-ws）

import httpx
from httpx_ws import connect_ws

async def websocket_example():
    async with httpx.AsyncClient() as client:
        async with connect_ws('wss://echo.websocket.org', client) as websocket:
            await websocket.send_text('Hello, WebSocket!')
            message = await websocket.receive_text()
            print(f"收到: {message}")

实用代码片段

API客户端类

import httpx
from typing import Optional, Dict, Any

class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.client = httpx.Client(
            base_url=base_url,
            headers={'Authorization': f'Bearer {api_key}'},
            timeout=30.0
        )

    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict[Any, Any]:
        response = self.client.get(endpoint, params=params)
        response.raise_for_status()
        return response.json()

    def post(self, endpoint: str, data: Dict[Any, Any]) -> Dict[Any, Any]:
        response = self.client.post(endpoint, json=data)
        response.raise_for_status()
        return response.json()

    def close(self):
        self.client.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

# 使用
with APIClient('https://api.example.com', 'your-api-key') as client:
    users = client.get('/users')
    new_user = client.post('/users', {'name': 'John', 'email': 'john@example.com'})

异步API客户端

import httpx
from typing import Optional, Dict, Any

class AsyncAPIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={'Authorization': f'Bearer {api_key}'},
            timeout=30.0
        )

    async def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict[Any, Any]:
        response = await self.client.get(endpoint, params=params)
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self.client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

# 使用
async def main():
    async with AsyncAPIClient('https://api.example.com', 'your-api-key') as client:
        users = await client.get('/users')
        print(users)

要求

httpx>=0.24.0
# 可选依赖项用于附加功能：
# httpx-ws>=0.6.0  # WebSocket支持
# tqdm>=4.65.0     # 进度条
# anyio>=3.7.0     # 替代异步运行时
# trio>=0.22.0     # 替代异步运行时

主要特性

• 同步和异步API：同步和异步代码使用相同的接口
• HTTP/2支持：具有多路复用的完整HTTP/2支持
• 连接池：高效的连接管理
• 流式传输：流式传输请求和响应，无需将所有内容加载到内存中
• 身份验证：内置支持基本、摘要、Bearer令牌和自定义身份验证
• 超时：可配置的连接、读取和整体请求超时
• 重定向处理：可配置的重定向跟随
• Cookie处理：自动Cookie管理
• 代理支持：HTTP和HTTPS代理支持
• SSL/TLS：完整的SSL/TLS配置选项

安装

uv add httpx

# 用于HTTP/2支持
uv add httpx[http2]

# 用于WebSocket支持
uv add httpx-ws

此技能为现代Python应用程序提供了全面的HTTP客户端功能，具有出色的异步支持和生产就绪的特性。