Python Snippets

Asynchronous HTTP Client with Retry Logic and Timeout Handling

import asyncio
import aiohttp
import time
from typing import Optional, Dict, Any
import logging

class AsyncHTTPClient:
    def __init__(self, timeout: int = 30, max_retries: int = 3, retry_delay: float = 1.0):
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.session: Optional[aiohttp.ClientSession] = None
        
    async def __aenter__(self):
        self.session = aiohttp.ClientSession(timeout=self.timeout)
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
            
    async def fetch(self, url: str, method: str = 'GET', **kwargs) -> Dict[str, Any]:
        """
        Fetch data from a URL with retry logic and error handling.
        
        Args:
            url: The URL to fetch
            method: HTTP method (GET, POST, PUT, DELETE, etc.)
            **kwargs: Additional arguments to pass to the request
            
        Returns:
            Dictionary with response data or error information
        """
        for attempt in range(self.max_retries + 1):
            try:
                if not self.session:
                    raise RuntimeError("Client not initialized. Use async context manager.")
                    
                start_time = time.time()
                async with self.session.request(method, url, **kwargs) as response:
                    response_time = time.time() - start_time
                    
                    result = {
                        'status': response.status,
                        'url': str(response.url),
                        'method': method,
                        'response_time': response_time,
                        'headers': dict(response.headers),
                        'success': True
                    }
                    
                    # Try to get JSON content first, fallback to text
                    try:
                        result['data'] = await response.json()
                    except aiohttp.ContentTypeError:
                        result['data'] = await response.text()
                        
                    return result
                    
            except asyncio.TimeoutError:
                if attempt < self.max_retries:
                    logging.warning(f"Timeout for {url} (attempt {attempt + 1}/{self.max_retries})")
                    await asyncio.sleep(self.retry_delay * (2 ** attempt))  # Exponential backoff
                else:
                    return {
                        'success': False,
                        'error': 'TimeoutError',
                        'message': f'Request to {url} timed out after {self.max_retries} retries',
                        'url': url,
                        'method': method
                    }
                    
            except aiohttp.ClientError as e:
                if attempt < self.max_retries:
                    logging.warning(f"Client error for {url}: {str(e)} (attempt {attempt + 1}/{self.max_retries})")
                    await asyncio.sleep(self.retry_delay * (2 ** attempt))
                else:
                    return {
                        'success': False,
                        'error': type(e).__name__,
                        'message': str(e),
                        'url': url,
                        'method': method
                    }
                    
            except Exception as e:
                return {
                    'success': False,
                    'error': type(e).__name__,
                    'message': str(e),
                    'url': url,
                    'method': method
                }
                
        # This should never be reached but included for completeness
        return {
            'success': False,
            'error': 'UnknownError',
            'message': 'An unknown error occurred',
            'url': url,
            'method': method
        }

# Example usage
async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/status/200',
        'https://httpbin.org/json',
        'https://httpbin.org/status/404',
    ]
    
    async with AsyncHTTPClient(timeout=10, max_retries=2) as client:
        tasks = [client.fetch(url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for result in results:
            if result['success']:
                print(f"✓ {result['status']} {result['url']} ({result['response_time']:.2f}s)")
            else:
                print(f"✗ {result['error']} {result['url']}: {result['message']}")

if __name__ == "__main__":
    # Configure logging
    logging.basicConfig(level=logging.WARNING)
    asyncio.run(main())

What This Code Does

This Python snippet implements a robust asynchronous HTTP client with built-in retry logic, timeout handling, and error management. The key features include:

Asynchronous Operation: Uses aiohttp for non-blocking HTTP requests, allowing multiple requests to be processed concurrently
Retry Logic with Exponential Backoff: Automatically retries failed requests with increasing delays between attempts
Configurable Timeouts: Adjustable timeout values to prevent requests from hanging indefinitely
Comprehensive Error Handling: Distinguishes between different types of errors (timeouts, connection errors, etc.) and provides detailed error information
Flexible HTTP Methods: Supports all standard HTTP methods (GET, POST, PUT, DELETE, etc.)
Response Time Measurement: Tracks how long each request takes to complete
Automatic Content Handling: Attempts to parse JSON responses automatically, falling back to text if needed
Context Manager Support: Ensures proper resource cleanup through async context managers

Why This Code Is Useful

Making HTTP requests in modern applications often requires handling unreliable network conditions, rate limits, and various error scenarios. This client solves these common problems:

Network Resilience: Real-world networks are unreliable. The retry mechanism handles temporary failures gracefully
Performance: Asynchronous operation allows processing multiple requests concurrently without blocking
Resource Management: Proper session management prevents resource leaks
Observability: Detailed response information helps with debugging and monitoring
Flexibility: Configurable parameters make it suitable for different use cases

How to Run the Code

Install the required dependency:
```
pip install aiohttp
```
Save the code to a file (e.g., async_client.py)
Run the example:
```
python async_client.py
```

To use the client in your own code:

import asyncio

async def example():
    async with AsyncHTTPClient(timeout=15, max_retries=3) as client:
        # Simple GET request
        result = await client.fetch('https://api.example.com/data')
        
        if result['success']:
            print(f"Data: {result['data']}")
        else:
            print(f"Error: {result['message']}")
            
        # POST request with JSON data
        post_result = await client.fetch(
            'https://api.example.com/users',
            method='POST',
            json={'name': 'John', 'email': 'john@example.com'}
        )

asyncio.run(example())

The client will automatically handle retries, timeouts, and errors while providing detailed information about each request’s outcome.