httpx
Installation
SKILL.md
HTTPX Skill
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.
Quick Start
Basic Usage
import httpx
# Simple GET request
response = httpx.get('https://api.example.com/data')
print(response.status_code)
print(response.json())
# POST request with JSON data
response = httpx.post('https://api.example.com/users', json={'name': 'Alice'})
Async Usage
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
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
import httpx
# Basic Authentication
auth = httpx.BasicAuth(username='user', password='pass')
client = httpx.Client(auth=auth)
# Bearer Token Authentication
headers = {'Authorization': 'Bearer your-token-here'}
client = httpx.Client(headers=headers)
# Per-request authentication
response = client.get('https://api.example.com', auth=('user', 'pass'))
3. Streaming Downloads
import httpx
# 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)
# 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)
4. Streaming Uploads
import httpx
# 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
5. Error Handling and Timeouts
import httpx
# 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}")
6. Client Configuration
import httpx
# 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
)
# All requests will use base_url and headers
response = client.get('/users') # Makes request to https://api.example.com/users
7. Custom Authentication
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
auth = CustomAuth('your-secret-api-key')
client = httpx.Client(auth=auth)
8. Progress Monitoring
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
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)
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
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
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'})
Async API Client
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
async def main():
async with AsyncAPIClient('https://api.example.com', 'your-api-key') as client:
users = await client.get('/users')
print(users)
Requirements
httpx>=0.24.0
# Optional dependencies for additional features:
# httpx-ws>=0.6.0 # WebSocket support
# tqdm>=4.65.0 # Progress bars
# anyio>=3.7.0 # Alternative async runtime
# trio>=0.22.0 # Alternative async runtime
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
Installation
uv add httpx
# For HTTP/2 support
uv add httpx[http2]
# For WebSocket support
uv add httpx-ws
This skill provides comprehensive HTTP client capabilities for modern Python applications, with excellent async support and production-ready features.
Related skills
More from bossjones/boss-file-utils
pydantic
Data validation and settings management using Python type annotations with Pydantic v2
1pytest
Python testing framework for writing simple, scalable, and powerful tests
1alembic
Database migration management for SQLAlchemy projects using Alembic
1uvicorn
ASGI server for Python web applications - Fast, production-ready server for async frameworks
1apscheduler
Advanced Python Scheduler - Task scheduling and job queue system
1docker-workflow
Comprehensive Docker containerization workflow covering multi-stage builds, docker-compose orchestration, image optimization, debugging, and production best practices. Use when containerizing applications, setting up development environments, or deploying with Docker.
1