本文へ移動
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(原文)

インストールする前に、エージェントに与えられる指示の中身を確認できます。

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

レビュー

まだレビューはありません。使ってみた感想をお寄せください。

同じリポジトリのスキル

概要と使いどころ

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-skills222026年10月9日 更新

ast-grep

無料

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

日本語の概要は準備中です。原文の説明を表示しています。

CodeAtCode/oss-ai-skills222026年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-skills222026年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-skills222026年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-skills222026年10月9日 更新

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

日本語の概要は準備中です。原文の説明を表示しています。

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

CodeAtCode のスキルをすべて見る

このスキルの問題を報告する