Stackvora.tech

Async HTTP Requests with httpx

Making synchronous HTTP requests in Python, especially when dealing with multiple external services or long-polling operations, can lead to significant…

Making synchronous HTTP requests in Python, especially when dealing with multiple external services or long-polling operations, can lead to significant performance bottlenecks due to I/O blocking. The httpx library provides a modern, fully-featured asynchronous HTTP client that integrates seamlessly with Python's asyncio framework, enabling non-blocking network operations. This allows applications to maintain responsiveness and efficiently handle concurrent requests without the overhead of threads or multiprocessing for I/O-bound tasks.

This article explores the core capabilities of httpx for asynchronous HTTP requests, covering client initialization, request execution, response handling, and advanced features like connection pooling and authentication. We'll demonstrate practical usage patterns with code examples, focusing on performance and robust error handling.

Getting Started with httpx Async

Before diving into examples, ensure you have httpx installed. It's a pure Python library and can be installed via pip:

pip install httpx

The fundamental building block for asynchronous requests with httpx is the AsyncClient. This client manages connection pools, cookies, and other session-related state, making it ideal for making multiple requests to the same or different hosts efficiently.

Basic Asynchronous GET Request

Here's a simple example demonstrating how to make an asynchronous GET request and process its JSON response using httpx.AsyncClient within an asyncio event loop.

import asyncio
import httpx

async def fetch_json(url: str) -> dict:
    """Fetches a URL and returns its JSON response."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(url)
            response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
            return response.json()
        except httpx.RequestError as exc:
            print(f"An error occurred while requesting {exc.request.url!r}: {exc}")
            raise # Re-raise for calling context to handle
        except httpx.HTTPStatusError as exc:
            print(f"Error response {exc.response.status_code} while requesting {exc.request.url!r}: {exc}")
            raise

async def main():
    urls = [
        "https://api.github.com/users/octocat",
        "https://httpbin.org/get?param=async-test",
        "https://api.github.com/nonexistent-endpoint" # This will likely raise a 404
    ]
    
    print("Fetching multiple URLs concurrently...")
    
    # Use asyncio.gather to run multiple coroutines concurrently
    # return_exceptions=True allows all tasks to complete, even if some fail.
    results = await asyncio.gather(*(fetch_json(url) for url in urls), return_exceptions=True)
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Request to {urls[i]} failed: {result}")
        else:
            print(f"Request to {urls[i]} successful. Partial data: {str(result)[:100]}...")

if __name__ == "__main__":
    asyncio.run(main())

In this example, httpx.AsyncClient(timeout=10.0) initializes a client with a 10-second timeout for the entire request (connection + read). The async with statement ensures the client's resources are properly cleaned up upon exiting the block, closing idle connections in the pool. response.raise_for_status() is a crucial method that checks the HTTP status code and raises httpx.HTTPStatusError for 4xx or 5xx responses, simplifying error handling.

Request Methods and Parameters

httpx supports all standard HTTP methods (GET, POST, PUT, DELETE, PATCH, OPTIONS, HEAD) with a consistent API. Parameters can be passed in various ways:

  • Query parameters: Use the params argument for URL query strings.
  • Request body:
    • json: For JSON-encoded body (automatically sets Content-Type: application/json).
    • data: For form-encoded body (automatically sets Content-Type: application/x-www-form-urlencoded).
    • files: For multipart form data (e.g., file uploads).
    • content: For raw bytes or string body (specify headers={"Content-Type": "text/plain"} manually).
  • Headers: Use the headers argument, a dictionary of string key-value pairs.
  • Timeouts: Can be set per-request or globally on the AsyncClient.

Example: POST Request with JSON Body and Custom Headers

async def create_resource(data: dict, auth_token: str) -> dict:
    """Sends a POST request with JSON data and authorization header."""
    url = "https://httpbin.org/post"
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "X-Custom-Header": "MyAsyncApp"
    }
    
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(url, json=data, headers=headers)
        response.raise_for_status()
        return response.json()

async def post_example():
    payload = {"name": "Async Item", "value": 123}
    token = "your_secret_token_123" # In a real app, retrieve securely
    
    try:
        new_item = await create_resource(payload, token)
        print(f"Successfully created item: {new_item['json']}")
        print(f"Headers sent: {new_item['headers']['Authorization']}")
    except httpx.HTTPStatusError as exc:
        print(f"Failed to create resource: {exc.response.status_code} - {exc.response.text}")
    except httpx.RequestError as exc:
        print(f"Network error during POST: {exc}")

if __name__ == "__main__":
    asyncio.run(post_example())

Note how httpx automatically handles the Content-Type header when json=data is used. The httpbin.org/post endpoint is excellent for inspecting what the server receives.

Advanced Features and Best Practices

Client-level Configuration and Connection Pooling

httpx.AsyncClient is designed for reusability. Initializing it once and reusing it across multiple requests (ideally for the lifetime of your application or a significant portion of its execution) allows for efficient connection pooling, TLS session reuse, and consistent settings.

# Global client instance for an application (e.g., in a FastAPI app startup)
# Or for a long-running background task
global_async_client = httpx.AsyncClient(
    base_url="https://api.example.com/v1/",
    headers={"User-Agent": "MyAwesomeAsyncApp/1.0"},
    timeout=httpx.Timeout(10.0, connect=5.0), # Total timeout, with a specific connect timeout
    follow_redirects=True,
    verify=True # Enable SSL verification
)

async def setup_client():
    # Perform any initial setup if needed
    print("AsyncClient initialized globally.")

async def shutdown_client():
    await global_async_client.aclose()
    print("AsyncClient closed globally.")

async def fetch_user(user_id: str) -> dict:
    # Requests will automatically use base_url and global headers
    response = await global_async_client.get(f"/users/{user_id}")
    response.raise_for_status()
    return response.json()

async def use_global_client_example():
    await setup_client() # Imagine this runs on application startup
    try:
        user_data = await fetch_user("123")
        print(f"Fetched user data: {user_data}")
    finally:
        await shutdown_client() # Imagine this runs on application shutdown

if __name__ == "__main__":
    asyncio.run(use_global_client_example())

Using a global or long-lived AsyncClient is a best practice. Remember to call await client.aclose() when the client is no longer needed to properly release network resources.

Authentication

httpx supports various authentication methods, including Basic Auth and custom authentication schemes. For basic auth, use the auth parameter with a tuple (username, password):

from httpx import BasicAuth

async def fetch_protected_data(username: str, password: str):
    auth = BasicAuth(username, password)
    async with httpx.AsyncClient() as client:
        response = await client.get("https://httpbin.org/basic-auth/user/passwd", auth=auth)
        response.raise_for_status()
        print(f"Auth successful: {response.json()}")

# Example for custom token auth (already shown implicitly with headers)
async def fetch_with_token(token: str):
    headers = {"Authorization": f"Bearer {token}"}
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data", headers=headers)
        response.raise_for_status()
        return response.json()

Streaming Responses

For very large responses, reading the entire content into memory at once can be inefficient or lead to memory issues. httpx supports streaming responses:

async def stream_large_file(url: str, output_path: str):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url, follow_redirects=True, timeout=30.0) as response:
            response.raise_for_status()
            with open(output_path, "wb") as f:
                async for chunk in response.aiter_bytes():
                    f.write(chunk)
            print(f"Successfully streamed {url} to {output_path}")

async def stream_example():
    # Example: A large file from a public source or a test endpoint
    large_file_url = "https://speed.hetzner.de/100MB.bin" # A 100MB test file
    await stream_large_file(large_file_url, "downloaded_large_file.bin")

if __name__ == "__main__":
    asyncio.run(stream_example())

Using client.stream("GET", ...) and iterating with async for chunk in response.aiter_bytes() allows processing the response data in chunks, ideal for large downloads or server-sent events.

Common Pitfalls and Troubleshooting

  • Forgetting await: In asynchronous Python, operations like client.get() and response.json() are coroutines and must be awaited. Forgetting await will result in a coroutine object being returned, not the actual result, often leading to subtle bugs or TypeErrors.
  • Not closing AsyncClient: If you initialize AsyncClient outside an async with block, remember to explicitly call await client.aclose() when done to release resources and prevent connection leaks. Long-running applications should manage client lifecycle carefully.
  • Inadequate Timeouts: Default timeouts (or lack thereof) can cause requests to hang indefinitely. Always set explicit timeouts on the client or per-request using httpx.Timeout to prevent resource exhaustion and improve resilience. A good practice is to set a connect timeout and a separate read or total timeout.
  • Synchronous calls within async def: Mixing blocking (synchronous) I/O calls within an async def function can block the entire event loop, defeating the purpose of asynchronous programming. Ensure all I/O operations (including file I/O or database calls) are either asynchronous or offloaded to a thread pool (e.g., using loop.run_in_executor).
  • Handling DNS resolution: By default, httpx relies on the system's asynchronous DNS resolver. In some environments (e.g., containers with custom DNS setups), you might encounter DNS resolution issues. While httpx doesn't directly configure DNS servers, ensuring your container/OS has a robust resolver is key.

Back to the knowledge base · Ask the AI assistant