Back to Details

httpx

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

HTTPX Skill

HTTPX 使用指南

HTTPX is a fully featured HTTP client for Python that provides both synchronous and asynchronous APIs, with support for HTTP/1.1 and HTTP/2.
HTTPX是一个功能完整的Python HTTP客户端，提供同步和异步API，支持HTTP/1.1和HTTP/2协议。

Quick Start

快速开始

Basic Usage

基础用法

python
import httpx
python
import httpx

Simple GET request

Simple GET request

response = httpx.get('https://api.example.com/data') print(response.status_code) print(response.json())
response = httpx.get('https://api.example.com/data') print(response.status_code) print(response.json())

POST request with JSON data

POST request with JSON data

response = httpx.post('https://api.example.com/users', json={'name': 'Alice'})
undefined
response = httpx.post('https://api.example.com/users', json={'name': 'Alice'})
undefined

Async Usage

异步用法

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

Common Patterns

常见使用模式

1. Async Client with Connection Pooling

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

python
import httpx

async def make_multiple_requests():
    async with httpx.AsyncClient() as client:
        # Reuse the same client for multiple requests
        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]
python
import httpx

async def make_multiple_requests():
    async with httpx.AsyncClient() as client:
        # Reuse the same client for multiple requests
        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. Authentication

2. 身份验证

python
import httpx
python
import httpx

Basic Authentication

Basic Authentication

auth = httpx.BasicAuth(username='user', password='pass') client = httpx.Client(auth=auth)
auth = httpx.BasicAuth(username='user', password='pass') client = httpx.Client(auth=auth)

Bearer Token Authentication

Bearer Token Authentication

headers = {'Authorization': 'Bearer your-token-here'} client = httpx.Client(headers=headers)
headers = {'Authorization': 'Bearer your-token-here'} client = httpx.Client(headers=headers)

Per-request authentication

Per-request authentication

response = client.get('https://api.example.com', auth=('user', 'pass'))
undefined
response = client.get('https://api.example.com', auth=('user', 'pass'))
undefined

3. Streaming Downloads

3. 流式下载

python
import httpx
python
import httpx

Stream large files without loading into memory

Stream large files without loading into memory

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)
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 streaming

Async streaming

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)
undefined
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)
undefined

4. Streaming Uploads

4. 流式上传

python
import httpx
python
import httpx

Upload large files with streaming

Upload large files with streaming

async def upload_large_file(): def generate_data(): # Yield chunks of data for i in range(100): yield f'chunk {i}\n'.encode()
async with httpx.AsyncClient() as client:
    response = await client.post(
        'https://api.example.com/upload',
        content=generate_data()
    )
    return response
undefined
async def upload_large_file(): def generate_data(): # Yield chunks of data for i in range(100): yield f'chunk {i}\n'.encode()
async with httpx.AsyncClient() as client:
    response = await client.post(
        'https://api.example.com/upload',
        content=generate_data()
    )
    return response
undefined

5. Error Handling and Timeouts

5. 错误处理与超时设置

python
import httpx
python
import httpx

Configure timeouts

Configure timeouts

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() # Raise exception for 4XX/5XX responses except httpx.TimeoutException: print("Request timed out") except httpx.HTTPStatusError as e: print(f"HTTP error: {e.response.status_code}") except httpx.RequestError as e: print(f"Request failed: {e}")
undefined
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() # Raise exception for 4XX/5XX responses except httpx.TimeoutException: print("Request timed out") except httpx.HTTPStatusError as e: print(f"HTTP error: {e.response.status_code}") except httpx.RequestError as e: print(f"Request failed: {e}")
undefined

6. Client Configuration

6. 客户端配置

python
import httpx
python
import httpx

Client with shared configuration

Client with shared configuration

client = httpx.Client( base_url='https://api.example.com', headers={'User-Agent': 'MyApp/1.0'}, timeout=30.0, follow_redirects=True )
client = httpx.Client( base_url='https://api.example.com', headers={'User-Agent': 'MyApp/1.0'}, timeout=30.0, follow_redirects=True )

All requests will use base_url and headers

All requests will use base_url and headers

response = client.get('/users') # Makes request to https://api.example.com/users
undefined
response = client.get('/users') # Makes request to https://api.example.com/users
undefined

7. Custom Authentication

7. 自定义身份验证

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

Use custom auth

Use custom auth

auth = CustomAuth('your-secret-api-key') client = httpx.Client(auth=auth)
undefined
auth = CustomAuth('your-secret-api-key') client = httpx.Client(auth=auth)
undefined

8. Progress Monitoring

8. 进度监控

python
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))
python
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. Retry Logic

9. 重试逻辑

python
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)  # Exponential backoff
python
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)  # Exponential backoff

10. WebSocket Support (with httpx-ws)

10. WebSocket 支持（需安装 httpx-ws）

python
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"Received: {message}")
python
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"Received: {message}")

Practical Code Snippets

实用代码片段

API Client Class

API 客户端类

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

Usage

Usage

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'})
undefined
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'})
undefined

Async API Client

异步 API 客户端

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

Usage

Usage

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

Requirements

依赖要求

txt
httpx>=0.24.0
txt
httpx>=0.24.0

Optional dependencies for additional features:

Optional dependencies for additional features:

httpx-ws>=0.6.0 # WebSocket support

httpx-ws>=0.6.0 # WebSocket support

tqdm>=4.65.0 # Progress bars

tqdm>=4.65.0 # Progress bars

anyio>=3.7.0 # Alternative async runtime

anyio>=3.7.0 # Alternative async runtime

trio>=0.22.0 # Alternative async runtime

trio>=0.22.0 # Alternative async runtime

undefined
undefined

Key Features

核心特性

  • Sync and Async APIs: Same interface for both synchronous and asynchronous code
  • HTTP/2 Support: Full HTTP/2 support with multiplexing
  • Connection Pooling: Efficient connection management
  • Streaming: Stream requests and responses without loading everything into memory
  • Authentication: Built-in support for Basic, Digest, Bearer token, and custom auth
  • Timeouts: Configurable timeouts for connect, read, and overall requests
  • Redirect Handling: Configurable redirect following
  • Cookie Handling: Automatic cookie management
  • Proxy Support: HTTP and HTTPS proxy support
  • SSL/TLS: Full SSL/TLS configuration options
  • 同步与异步API: 同步和异步代码使用相同的接口
  • HTTP/2支持: 完整的HTTP/2支持，支持多路复用
  • 连接池: 高效的连接管理
  • 流式处理: 无需将全部内容加载到内存即可流式处理请求和响应
  • 身份验证: 内置支持Basic、Digest、Bearer令牌以及自定义身份验证
  • 超时设置: 可配置连接、读取和整体请求的超时时间
  • 重定向处理: 可配置是否跟随重定向
  • Cookie处理: 自动管理Cookie
  • 代理支持: 支持HTTP和HTTPS代理
  • SSL/TLS: 完整的SSL/TLS配置选项

Installation

安装方法

bash
pip install httpx
bash
pip install httpx

For HTTP/2 support

如需HTTP/2支持

pip install httpx[http2]
pip install httpx[http2]

For WebSocket support

如需WebSocket支持

pip install httpx-ws

This skill provides comprehensive HTTP client capabilities for modern Python applications, with excellent async support and production-ready features.
pip install httpx-ws

本指南为现代Python应用程序提供了全面的HTTP客户端能力，具备出色的异步支持和生产级特性。