philandClaude
25086fc01b
- Modular architecture with separate modules for scraping, parsing, security, validation, and caching - Comprehensive security measures including HTML sanitization, rate limiting, and input validation - Robust error handling with custom exceptions and retry logic - HTTP caching with ETags and Last-Modified headers for efficiency - Pre-compiled regex patterns for improved performance - Comprehensive test suite with 66 tests covering all major functionality - Docker support for containerized deployment - Configuration management with environment variable support - Working parser that successfully extracts 32 articles from Warhammer Community 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
"""Retry utilities with exponential backoff for network operations."""
|
|
|
|
import time
|
|
import logging
|
|
from typing import Any, Callable, Optional, Type, Union, Tuple
|
|
from functools import wraps
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RetryConfig:
|
|
"""Configuration for retry behavior."""
|
|
|
|
def __init__(
|
|
self,
|
|
max_attempts: int = 3,
|
|
base_delay: float = 1.0,
|
|
max_delay: float = 60.0,
|
|
backoff_factor: float = 2.0,
|
|
jitter: bool = True
|
|
):
|
|
self.max_attempts = max_attempts
|
|
self.base_delay = base_delay
|
|
self.max_delay = max_delay
|
|
self.backoff_factor = backoff_factor
|
|
self.jitter = jitter
|
|
|
|
|
|
def calculate_delay(attempt: int, config: RetryConfig) -> float:
|
|
"""Calculate delay for retry attempt with exponential backoff."""
|
|
delay = config.base_delay * (config.backoff_factor ** (attempt - 1))
|
|
delay = min(delay, config.max_delay)
|
|
|
|
if config.jitter:
|
|
# Add random jitter to avoid thundering herd
|
|
import random
|
|
jitter_amount = delay * 0.1
|
|
delay += random.uniform(-jitter_amount, jitter_amount)
|
|
|
|
return max(0, delay)
|
|
|
|
|
|
def retry_on_exception(
|
|
exceptions: Union[Type[Exception], Tuple[Type[Exception], ...]],
|
|
config: Optional[RetryConfig] = None
|
|
) -> Callable:
|
|
"""Decorator to retry function calls on specific exceptions.
|
|
|
|
Args:
|
|
exceptions: Exception type(s) to retry on
|
|
config: Retry configuration, uses default if None
|
|
|
|
Returns:
|
|
Decorated function with retry logic
|
|
"""
|
|
if config is None:
|
|
config = RetryConfig()
|
|
|
|
def decorator(func: Callable) -> Callable:
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs) -> Any:
|
|
last_exception = None
|
|
|
|
for attempt in range(1, config.max_attempts + 1):
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
if attempt > 1:
|
|
logger.info(f"{func.__name__} succeeded on attempt {attempt}")
|
|
return result
|
|
|
|
except exceptions as e:
|
|
last_exception = e
|
|
|
|
if attempt == config.max_attempts:
|
|
logger.error(
|
|
f"{func.__name__} failed after {config.max_attempts} attempts. "
|
|
f"Final error: {e}"
|
|
)
|
|
raise
|
|
|
|
delay = calculate_delay(attempt, config)
|
|
logger.warning(
|
|
f"{func.__name__} attempt {attempt} failed: {e}. "
|
|
f"Retrying in {delay:.2f} seconds..."
|
|
)
|
|
time.sleep(delay)
|
|
|
|
except Exception as e:
|
|
# Don't retry on unexpected exceptions
|
|
logger.error(f"{func.__name__} failed with unexpected error: {e}")
|
|
raise
|
|
|
|
# This should never be reached, but just in case
|
|
if last_exception:
|
|
raise last_exception
|
|
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
# Common retry configurations for different scenarios
|
|
NETWORK_RETRY_CONFIG = RetryConfig(
|
|
max_attempts=3,
|
|
base_delay=1.0,
|
|
max_delay=30.0,
|
|
backoff_factor=2.0,
|
|
jitter=True
|
|
)
|
|
|
|
PLAYWRIGHT_RETRY_CONFIG = RetryConfig(
|
|
max_attempts=2,
|
|
base_delay=2.0,
|
|
max_delay=10.0,
|
|
backoff_factor=2.0,
|
|
jitter=False
|
|
)
|
|
|
|
FILE_RETRY_CONFIG = RetryConfig(
|
|
max_attempts=3,
|
|
base_delay=0.5,
|
|
max_delay=5.0,
|
|
backoff_factor=1.5,
|
|
jitter=False
|
|
) |