跳到正文
cccskills
免费在 GitHub 公开

httpx

Use when making HTTP requests in Python with httpx - sync and async clients, connection pooling, timeouts, HTTP/2, retry patterns, transport configuration, or choosing between httpx/requests/aiohttp

中文简介正在准备中。当前显示原文说明。

查看安装方法

包含的文件(4)

  • SKILL.md11.5 KB
  • references/advanced.md3.5 KB
  • references/async.md3.0 KB
  • references/testing-integration.md4.6 KB

SKILL.md(原文)

安装前可以查看交给 AI 代理的指令内容。

httpx

Modern HTTP client with sync/async APIs, HTTP/2, and production-grade features.

Quick Start

import httpx

# Always use timeouts in production
timeout = httpx.Timeout(
    connect=5.0,   # Connection establishment
    read=30.0,     # Response body read
    write=10.0,    # Request body write
    pool=5.0       # Connection pool acquisition
)

with httpx.Client(timeout=timeout) as client:
    response = client.get("https://api.example.com/data")
    print(response.status_code)
    print(response.json())

See Quick Start for basic request patterns.

Production Configuration

Timeouts (Mandatory)

The most common production failure is httpx.get(url) with no timeout—a hung server holds the connection forever.

import httpx

# Configure once on the client, reuse across requests
timeout = httpx.Timeout(
    connect=5.0,   # Fails if TCP handshake exceeds this
    read=30.0,     # Fails if server stops sending data
    write=10.0,    # Fails if client can't send request
    pool=5.0       # Fails if no connection available in pool
)

client = httpx.Client(timeout=timeout)

Timeout exception hierarchy:

ExceptionWhen it firesWhat it indicates
ConnectTimeoutTCP handshake / TLS handshakeServer unreachable or firewall blocking
ReadTimeoutWaiting for response bodyServer processing too slowly
WriteTimeoutSending request bodyClient network saturated
PoolTimeoutWaiting for connection from poolConnection pool exhausted
from httpx import ConnectTimeout, ReadTimeout, TimeoutException

try:
    response = client.get("https://api.example.com")
except ConnectTimeout:
    # Server unreachable—retry with backoff or fail fast
    pass
except ReadTimeout:
    # Server hung—may have partial response
    pass
except TimeoutException as e:
    # Catch-all for any timeout subclass
    print(f"Timeout: {e}")

Connection Pooling

Rule: One httpx.Client per application/process, reused across requests. A client per request defeats pooling and leaks connections.

# Correct: Single client reused
class APIService:
    def __init__(self):
        limits = httpx.Limits(
            max_connections=50,          # Max total connections
            max_keepalive_connections=20, # Max idle connections
            keepalive_expiry=30          # Keepalive timeout (seconds)
        )
        timeout = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)
        self.client = httpx.Client(limits=limits, timeout=timeout)

    def fetch_user(self, user_id: int):
        return self.client.get(f"/users/{user_id}")

    def shutdown(self):
        self.client.close()  # Call on app shutdown hook

Common pooling mistakes:

# Wrong: New client per request (leaks connections)
def handle_request():
    with httpx.Client() as client:  # Creates new connection pool
        return client.get("https://api.example.com")

# Wrong: Sync and async clients are not interchangeable
async def fetch_async():
    with httpx.Client() as client:  # Blocks event loop!
        return client.get("https://api.example.com")

# Correct: Use AsyncClient in async contexts
async def fetch_async():
    async with httpx.AsyncClient() as client:
        return await client.get("https://api.example.com")

HTTP/2 Configuration

# Enable HTTP/2 (requires httpx[http2])
client = httpx.Client(http2=True)

# HTTP/2 requires ALPN negotiation
# If server doesn't support it:
# - Silent fallback to HTTP/1.1 (most common)
# - RemoteProtocolError if ALPN fails explicitly

When HTTP/2 is worth it:

  • ✅ Many concurrent requests to the same host (multiplexing benefit)
  • ✅ Large response bodies (header compression helps)
  • ❌ Small request counts (TLS handshake overhead dominates)
  • ❌ Single-request scripts (no multiplexing benefit)
# Check if HTTP/2 was negotiated
response = client.get("https://api.example.com")
print(response.http_version)  # "HTTP/2" or "HTTP/1.1"

Retry Patterns (Idempotency Only)

Never retry non-idempotent requests. A retried POST can duplicate a charge, create duplicate records, or trigger side effects twice.

import httpx
import time
import random

def exponential_backoff_with_jitter(attempt: int, base: float = 1.0, max_delay: float = 30.0) -> float:
    """Bounded exponential backoff with jitter."""
    delay = min(base * (2 ** attempt), max_delay)
    jitter = random.uniform(0, delay * 0.1)  # 10% jitter
    return delay + jitter

def safe_get_with_retries(client: httpx.Client, url: str, max_retries: int = 3):
    """Retry-safe GET request with exponential backoff."""
    for attempt in range(max_retries):
        try:
            response = client.get(url)
            if response.status_code >= 500:
                # Server error—may be safe to retry
                time.sleep(exponential_backoff_with_jitter(attempt))
                continue
            return response
        except (httpx.ConnectTimeout, httpx.ReadTimeout) as e:
            if attempt == max_retries - 1:
                raise  # Last attempt failed
            time.sleep(exponential_backoff_with_jitter(attempt))
    raise RuntimeError("Should not reach here")

# NEVER RETRY: Non-idempotent operations
def create_user(client: httpx.Client, data: dict):
    """Post is not retry-safe—do not wrap in retry logic."""
    return client.post("/users", json=data)
    # If this fails, you don't know if it succeeded or not
    # Implement idempotency keys instead of retries

Transport-level retries:

httpx doesn't include urllib3.Retry-style built-in retries because idempotency is context-dependent. Use transport mounts for fine-grained control:

# Custom retry transport (advanced)
class RetryTransport(httpx.BaseTransport):
    def __init__(self, transport: httpx.BaseTransport, max_retries: int = 3):
        self.transport = transport
        self.max_retries = max_retries

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        # Implement retry logic here, checking request.method for idempotency
        pass

client = httpx.Client(transport=RetryTransport(httpx.HTTPTransport()))

Client Selection Guide

Before choosing an HTTP client, run this check:

# Decision flow:
# 1. Does the codebase need async?
#    - Yes → httpx or aiohttp
#    - No → httpx or requests
#
# 2. Does it need HTTP/2?
#    - Yes → httpx (only option with HTTP/2)
#    - No → continue
#
# 3. Is dependency count a constraint?
#    - Yes → stdlib urllib.request
#    - No → continue
#
# 4. Sync or async?
#    - Sync → requests (ubiquitous) or httpx (modern typing)
#    - Async → httpx (sync+async) or aiohttp (async-only, server-side focus)
ClientSyncAsyncHTTP/2DependenciesBest for
requests✅❌❌1 (urllib3)Simple sync scripts, ubiquitous ecosystem
httpx✅✅✅2 (httpcore, certifi)Modern apps needing async or HTTP/2
aiohttp❌✅❌3 (aiohttp, yarl, multidict)Async servers (client + server in one)
urllib.request✅❌❌0 (stdlib)Zero-dependency scripts, constrained envs

Testing

Why respx beats monkeypatching internals: Mocking at the transport layer avoids implementation details, works with both sync and async clients, and doesn't break when httpx updates its internals.

import httpx
import respx
import pytest

@respx.mock
def test_request_method_and_headers():
    """Assert on request method, URL, and headers."""
    route = respx.post("https://api.example.com/users").mock(
        return_value=httpx.Response(201, json={"id": 123})
    )

    client = httpx.Client()
    response = client.post(
        "https://api.example.com/users",
        json={"name": "Alice"},
        headers={"X-Request-ID": "abc-123"}
    )

    assert response.status_code == 201
    assert response.json() == {"id": 123}

    # Assert request was made correctly
    assert route.called
    request = route.calls[0].request
    assert request.method == b"POST"
    assert "X-Request-ID" in request.headers

Testing retry behavior without wall-clock delay:

import httpx
from httpx import MockTransport

def test_retry_behavior():
    """Test retry logic by injecting a failing transport."""
    call_count = 0

    def failing_transport(request):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise httpx.ConnectTimeout("Simulated timeout", request=request)
        return httpx.Response(200, json={"success": True})

    transport = MockTransport(failing_transport)
    client = httpx.Client(transport=transport)

    # Your retry logic should handle this
    response = client.get("https://example.com")
    assert response.json() == {"success": True}
    assert call_count == 3  # Verified retry count

Testing async with respx:

import pytest
import httpx
import respx

@pytest.mark.asyncio
async def test_async_client():
    @respx.mock
    async def inner():
        route = respx.get("https://api.example.com/data").mock(
            return_value=httpx.Response(200, json={"data": "test"})
        )

        async with httpx.AsyncClient() as client:
            response = await client.get("https://api.example.com/data")
            assert response.json() == {"data": "test"}
            assert route.called

    await inner()

Common Pitfalls

Connection Leaks

# Wrong: Client not closed
client = httpx.Client()
response = client.get("https://example.com")
# Connection pool never cleaned up

# Correct: Context manager or explicit close
with httpx.Client() as client:
    response = client.get("https://example.com")

# Or for long-lived clients
client = httpx.Client()
try:
    response = client.get("https://example.com")
finally:
    client.close()

Mixing Sync and Async

# Wrong: Sync client in async function blocks event loop
async def fetch():
    with httpx.Client() as client:  # Blocks!
        return client.get("https://example.com")

# Correct: Use AsyncClient
async def fetch():
    async with httpx.AsyncClient() as client:
        return await client.get("https://example.com")

Timeout Not Set

# Wrong: No timeout—server can hang forever
httpx.get("https://slow-server.com")

# Correct: Explicit timeout
httpx.get("https://slow-server.com", timeout=httpx.Timeout(connect=5.0, read=30.0))

Deep Dives

Load these reference files on demand when you need deeper coverage:

  • Async patterns — references/async.md — Concurrent requests, streaming uploads/downloads, event loop integration
  • Advanced features — references/advanced.md — HTTP/2 deep dive, proxy configuration, event hooks, custom transports
  • Testing & framework integration — references/testing-integration.md — FastAPI/Django integration, advanced respx patterns, async test fixtures

References

评价

还没有评价。欢迎分享你的使用感受。

评论

无需登录即可发表。欢迎分享使用感受、提问,或说明在哪些环境下无法运行。

还没有评论。来写第一条吧。

提交后会立即公开。请勿填写邮箱、电话等个人信息。URL 最多 2 个。

0 / 1000 字

此表单受 reCAPTCHA 保护,适用 Google 的隐私权政策和服务条款。

同一仓库的技能

aiohttp

免费

Use when building Python async HTTP services or clients with aiohttp - web server routing, middleware, WebSocket, SSE, streaming, client sessions, pytest-aiohttp testing, or troubleshooting SSL and timeout issues

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

ast-grep

免费

Use when doing structural code search and rewriting - ast-grep linting, refactoring, multi-language patterns

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

Use when building GBA games with the BPCore Lua engine - entity, sprite and tilemap functions, SRAM save and load, link cable multiplayer protocol, camera and scrolling, or optimization patterns

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

celery

免费

Use when running background tasks with Celery - worker and broker configuration (Redis, RabbitMQ), task routing by name vs queue, chains/groups/chords, retry patterns (autoretry_for, retry_backoff), acks_late semantics, failure detection, and monitoring with Flower

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

django

免费

Use when building Django applications - security hardening, authentication and permissions, ORM optimization, PostgreSQL features, Django 6.0, migrations, testing, and ecosystem libraries

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

Use when customizing Django Admin - save_formset, get_search_results, formsets, queryset optimization, db_index, custom URLs

中文简介正在准备中。当前显示原文说明。

CodeAtCode/oss-ai-skills242026年10月9日 更新

查看 CodeAtCode 的全部技能

报告此技能的问题