Exceptions Reference¶
Complete guide to ProxyWhirl’s exception hierarchy, error codes, and error handling best practices.
Tip
All ProxyWhirl exceptions support structured error data via to_dict() and include a retry_recommended flag to help determine whether an operation should be retried. Use ProxyErrorCode enum values for reliable programmatic error handling.
Table of Contents¶
Exception Hierarchy¶
ProxyWhirl uses a hierarchical exception system with a common base class for all library-specific errors:
Exception
├── ProxyWhirlError (base)
│ ├── ProxyValidationError
│ ├── ProxyPoolEmptyError
│ ├── ProxyConnectionError
│ ├── ProxyAuthenticationError
│ ├── ProxyFetchError
│ ├── ProxyStorageError
│ ├── CacheCorruptionError
│ ├── CacheStorageError
│ ├── CacheValidationError (also inherits from ValueError)
│ └── RequestQueueFullError
├── RetryableError (retry module - triggers retry)
├── NonRetryableError (retry module - skips retry)
├── RegexTimeoutError (safe_regex module - ReDoS protection)
└── RegexComplexityError (safe_regex module - ReDoS protection)
Note
ProxyAuthenticationError inherits directly from ProxyWhirlError, not from ProxyConnectionError. Catch it separately from connection errors for proper credential handling.
All ProxyWhirl exceptions inherit from ProxyWhirlError, making it easy to catch all library-specific errors:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyWhirlError
rotator = ProxyWhirl()
try:
response = rotator.request("GET", "https://httpbin.org/ip")
except ProxyWhirlError as e:
# Catches ALL ProxyWhirl-specific errors
print(f"ProxyWhirl error: {e}")
print(f"Error code: {e.error_code}")
print(f"Retry recommended: {e.retry_recommended}")
ProxyErrorCode Enum¶
The ProxyErrorCode enum provides standardized error codes for programmatic error handling. All ProxyWhirl exceptions include an error_code attribute that maps to one of these codes.
from proxywhirl.exceptions import ProxyErrorCode
class ProxyErrorCode(str, Enum):
"""Error codes for programmatic error handling."""
PROXY_POOL_EMPTY = "PROXY_POOL_EMPTY"
PROXY_VALIDATION_FAILED = "PROXY_VALIDATION_FAILED"
PROXY_CONNECTION_FAILED = "PROXY_CONNECTION_FAILED"
PROXY_AUTH_FAILED = "PROXY_AUTH_FAILED"
PROXY_FETCH_FAILED = "PROXY_FETCH_FAILED"
PROXY_STORAGE_FAILED = "PROXY_STORAGE_FAILED"
CACHE_CORRUPTED = "CACHE_CORRUPTED"
CACHE_STORAGE_FAILED = "CACHE_STORAGE_FAILED"
CACHE_VALIDATION_FAILED = "CACHE_VALIDATION_FAILED"
TIMEOUT = "TIMEOUT"
NETWORK_ERROR = "NETWORK_ERROR"
INVALID_CONFIGURATION = "INVALID_CONFIGURATION"
QUEUE_FULL = "QUEUE_FULL"
Usage¶
Error codes enable reliable programmatic error handling without relying on string matching:
from proxywhirl.exceptions import ProxyWhirlError, ProxyErrorCode
try:
response = rotator.request("GET", "https://example.com")
except ProxyWhirlError as e:
# Match by error code (recommended)
if e.error_code == ProxyErrorCode.PROXY_POOL_EMPTY:
rotator.auto_fetch()
elif e.error_code == ProxyErrorCode.TIMEOUT:
response = rotator.request("GET", url, timeout=60)
# Also accessible as string value
print(f"Error code: {e.error_code.value}") # "PROXY_POOL_EMPTY"
Utility Functions¶
redact_url()¶
Important
ProxyWhirl automatically redacts sensitive information from URLs before including them in error messages. This prevents credentials and API keys from leaking into logs. You only need to call redact_url() manually when logging proxy URLs outside of exception handling.
from proxywhirl.exceptions import redact_url
def redact_url(url: str) -> str:
"""
Redact sensitive information from a URL.
Removes username and password while preserving scheme, host, port, and path.
Args:
url: URL to redact
Returns:
Redacted URL string
"""
Examples:
from proxywhirl.exceptions import redact_url
# Credentials are removed
original = "http://user:password@proxy.example.com:8080"
redacted = redact_url(original)
print(redacted) # "http://proxy.example.com:8080"
# Sensitive query parameters are masked
original = "https://api.example.com/data?token=secret123&key=abc"
redacted = redact_url(original)
print(redacted) # "https://api.example.com/data?token=***&key=***"
# Path and port are preserved
original = "socks5://admin:pass@proxy.example.com:1080/path"
redacted = redact_url(original)
print(redacted) # "socks5://proxy.example.com:1080/path"
Redacted Parameters:
Username and password (always removed)
Query parameters:
password,token,key,secret,auth(case-insensitive)
Security Note: All ProxyWhirl exceptions automatically redact URLs in the proxy_url attribute and error messages. You only need to call redact_url() manually when logging URLs outside of exception handling.
Error Codes¶
All exceptions include a programmatic error code for structured error handling:
Error Code |
Exception |
Description |
|---|---|---|
|
|
No proxies available in pool |
|
|
Invalid proxy URL or configuration |
|
|
Unable to connect through proxy |
|
|
Proxy authentication failed |
|
|
Failed to fetch proxies from source |
|
|
Storage operation failed |
|
|
Cache data is corrupted |
|
|
Cache backend unavailable |
|
|
Cache entry validation failed |
|
|
Request queue is full |
|
|
Request timeout (special case) |
|
|
Generic network error |
|
|
Invalid configuration |
Use error codes for programmatic error handling:
from proxywhirl.exceptions import ProxyWhirlError, ProxyErrorCode
try:
response = rotator.request("GET", "https://example.com")
except ProxyWhirlError as e:
if e.error_code == ProxyErrorCode.PROXY_POOL_EMPTY:
# Add more proxies
rotator.add_proxy("http://proxy.example.com:8080")
elif e.error_code == ProxyErrorCode.TIMEOUT:
# Increase timeout
rotator.request("GET", url, timeout=60)
elif e.error_code == ProxyErrorCode.PROXY_AUTH_FAILED:
# Update credentials
logger.error("Invalid proxy credentials")
Exception Reference¶
ProxyWhirlError¶
Base exception for all ProxyWhirl errors.
All ProxyWhirl exceptions inherit from this class and support rich metadata for debugging and retry logic.
Attributes¶
message(str): Human-readable error messageproxy_url(str | None): Redacted URL of the proxy that caused the errorerror_type(str | None): Type of error (e.g., “timeout”, “invalid_credentials”)error_code(ProxyErrorCode): Programmatic error code for handlingretry_recommended(bool): Whether retrying the operation is recommendedattempt_count(int | None): Number of attempts made before this errormetadata(dict): Additional error-specific metadata
Methods¶
def to_dict(self) -> dict[str, Any]:
"""Convert exception to dictionary for logging/serialization."""
Example¶
from proxywhirl.exceptions import ProxyWhirlError
try:
response = rotator.request("GET", "https://example.com")
except ProxyWhirlError as e:
# Access structured error data
error_dict = e.to_dict()
print(f"Error code: {error_dict['error_code']}")
print(f"Proxy URL: {error_dict['proxy_url']}") # Redacted for security
print(f"Retry: {error_dict['retry_recommended']}")
print(f"Attempt: {error_dict['attempt_count']}")
Security: URL Redaction¶
ProxyWhirl automatically redacts sensitive information from proxy URLs in error messages:
# Original URL: http://user:password@proxy.example.com:8080/path?token=secret
# Redacted URL: http://proxy.example.com:8080/path?token=***
# Credentials are always removed for security
ProxyValidationError¶
Raised when proxy URL or configuration is invalid.
Error Code:
PROXY_VALIDATION_FAILEDRetry Recommended: No (invalid input requires correction)
When Raised¶
Invalid proxy URL format
Unsupported protocol (not http, https, or socks5)
Malformed credentials
Invalid configuration parameters
Attributes¶
Same as ProxyWhirlError, plus automatic suggestion appended to message.
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyValidationError
rotator = ProxyWhirl()
try:
# Invalid URL format (missing scheme)
rotator.add_proxy("proxy.example.com:8080")
except ProxyValidationError as e:
print(e) # "Invalid proxy URL. Check proxy URL format and protocol."
# Fix: Add proper scheme
rotator.add_proxy("http://proxy.example.com:8080")
Resolution Steps¶
Verify proxy URL format:
protocol://host:portEnsure protocol is supported (http, https, socks5)
Check that credentials are properly URL-encoded
Validate port number is within valid range (1-65535)
ProxyPoolEmptyError¶
Raised when attempting to select from an empty proxy pool.
Error Code:
PROXY_POOL_EMPTYRetry Recommended: No (pool must be populated first)
When Raised¶
No proxies configured in the pool
All proxies filtered out by health checks
All circuit breakers are open (all proxies failing)
Geographic filtering excluded all proxies
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyPoolEmptyError
rotator = ProxyWhirl()
try:
# No proxies added yet
response = rotator.request("GET", "https://example.com")
except ProxyPoolEmptyError as e:
print(e) # "No proxies available in the pool. Add proxies using add_proxy()..."
# Solution 1: Add proxies manually
rotator.add_proxy("http://proxy1.example.com:8080")
rotator.add_proxy("http://proxy2.example.com:8080")
# Solution 2: Auto-fetch from sources
rotator.auto_fetch()
# Retry request
response = rotator.request("GET", "https://example.com")
Resolution Steps¶
Add proxies using
add_proxy()orauto_fetch()Check if proxies were filtered by health checks
Verify circuit breakers aren’t all open
Review geographic targeting settings if applicable
Check storage connection if using persistent storage
ProxyConnectionError¶
Raised when unable to connect through a proxy.
Error Code:
PROXY_CONNECTION_FAILED(orTIMEOUTfor timeout-specific errors)Retry Recommended: Yes (transient network issues may resolve)
Tip
When the error message contains “timeout”, the error_code is automatically set to ProxyErrorCode.TIMEOUT instead of PROXY_CONNECTION_FAILED. Check e.error_code to distinguish between timeout and other connection failures.
When Raised¶
Proxy is unreachable or offline
Network connectivity issues
Proxy doesn’t support target protocol
Request timeout exceeded
Circuit breaker is open for the proxy
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyConnectionError, ProxyErrorCode
rotator = ProxyWhirl()
rotator.add_proxy("http://unreachable.proxy.com:8080")
try:
response = rotator.request("GET", "https://example.com", timeout=5)
except ProxyConnectionError as e:
if e.error_code == ProxyErrorCode.TIMEOUT:
print("Request timed out - trying with longer timeout")
response = rotator.request("GET", "https://example.com", timeout=30)
else:
print(f"Connection failed: {e}")
# Proxy may be offline - try another one
rotator.remove_proxy("http://unreachable.proxy.com:8080")
Resolution Steps¶
Verify proxy is reachable (ping, telnet)
Check network connectivity and firewall rules
Ensure proxy supports the target protocol
Increase timeout value for slow proxies
Check circuit breaker status
Verify proxy provider isn’t blocking your IP
ProxyAuthenticationError¶
Raised when proxy authentication fails.
Error Code:
PROXY_AUTH_FAILEDRetry Recommended: No (credentials must be corrected)
When Raised¶
Invalid username or password (HTTP 401/407 response)
Credentials expired or revoked
Proxy requires different authentication method
IP whitelist restriction
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyAuthenticationError
rotator = ProxyWhirl()
try:
# Wrong credentials
rotator.add_proxy(
"http://proxy.example.com:8080",
username="user",
password="wrong_password"
)
response = rotator.request("GET", "https://example.com")
except ProxyAuthenticationError as e:
print(e) # "Proxy authentication failed (407)... Verify username and password..."
# Fix credentials
rotator.remove_proxy("http://proxy.example.com:8080")
rotator.add_proxy(
"http://proxy.example.com:8080",
username="user",
password="correct_password"
)
response = rotator.request("GET", "https://example.com")
Resolution Steps¶
Verify username and password are correct
Check if credentials have expired
Ensure IP address is whitelisted (if required)
Contact proxy provider for credential verification
Check if proxy requires specific auth method (Basic, Digest, NTLM)
ProxyFetchError¶
Raised when fetching proxies from external sources fails.
Error Code:
PROXY_FETCH_FAILEDRetry Recommended: Yes (source may be temporarily unavailable)
When Raised¶
Source URL is unreachable
API credentials are invalid
Response format doesn’t match expectations
Rate limit exceeded
Malformed JSON/text response
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyFetchError
import time
rotator = ProxyWhirl()
try:
# Fetch from external source
rotator.auto_fetch()
except ProxyFetchError as e:
print(f"Failed to fetch proxies: {e}")
# Solution 1: Retry after delay (may be rate limited)
time.sleep(60)
rotator.auto_fetch()
# Solution 2: Add proxies manually as fallback
rotator.add_proxy("http://fallback-proxy.example.com:8080")
Resolution Steps¶
Verify source URL is accessible
Check API credentials (if required)
Review rate limits with provider
Validate response format matches expected schema
Check proxy provider status page
Use manual proxy addition as fallback
ProxyStorageError¶
Raised when proxy storage operations fail.
Error Code:
PROXY_STORAGE_FAILEDRetry Recommended: No (storage issue must be resolved)
When Raised¶
Insufficient file system permissions
Disk space exhausted
Storage path is not writable
Database connection failed
SQLite database is locked
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyStorageError
import os
try:
# Using read-only directory
rotator = ProxyWhirl(storage_path="/read-only/proxywhirl.db")
except ProxyStorageError as e:
print(f"Storage error: {e}")
# Solution: Use writable directory
storage_path = os.path.expanduser("~/.proxywhirl/proxies.db")
os.makedirs(os.path.dirname(storage_path), exist_ok=True)
rotator = ProxyWhirl(storage_path=storage_path)
Resolution Steps¶
Check file system permissions (
chmod,chown)Verify disk space is available (
df -h)Ensure storage path is writable
Check database connection (if using external storage)
Close other processes accessing the database
Use in-memory storage as temporary fallback
CacheCorruptionError¶
Raised when cache data is corrupted and cannot be recovered.
Error Code:
CACHE_CORRUPTEDRetry Recommended: No (cache must be cleared)
When Raised¶
Cache file is corrupted
Invalid cache format version
Disk errors or partial writes
Encryption key mismatch (if using encrypted cache)
Example¶
from proxywhirl.cache import CacheManager
from proxywhirl.exceptions import CacheCorruptionError
cache = CacheManager()
try:
data = cache.get("my_key")
except CacheCorruptionError as e:
print(f"Cache corrupted: {e}")
# Solution: Clear cache and reinitialize
cache.clear()
cache = CacheManager()
print("Cache cleared and reinitialized")
Resolution Steps¶
Clear the cache directory
Reinitialize cache system
Check for disk errors (
fsckon Linux)Verify cache format version compatibility
Ensure encryption key is consistent (if applicable)
CacheStorageError¶
Raised when cache storage backend is unavailable.
Error Code:
CACHE_STORAGE_FAILEDRetry Recommended: Yes (backend may recover)
When Raised¶
Redis/Memcached server is down
Network connectivity to cache server lost
Invalid cache credentials
Cache server out of memory
Connection pool exhausted
Example¶
from proxywhirl.cache import CacheManager
from proxywhirl.exceptions import CacheStorageError
import time
cache = CacheManager(backend="redis", redis_url="redis://localhost:6379")
try:
cache.set("key", "value")
except CacheStorageError as e:
print(f"Cache backend unavailable: {e}")
# Solution 1: Retry with exponential backoff
for i in range(3):
time.sleep(2 ** i)
try:
cache.set("key", "value")
break
except CacheStorageError:
continue
# Solution 2: Fallback to in-memory cache
cache = CacheManager(backend="memory")
Resolution Steps¶
Verify cache backend (Redis, Memcached) is running
Check network connectivity to cache server
Validate cache credentials
Review cache server logs for errors
Check server memory and resource limits
Use fallback cache (in-memory) temporarily
CacheValidationError¶
Raised when cache entry fails validation.
Error Code:
CACHE_VALIDATION_FAILEDRetry Recommended: No (data must be corrected)
Note: Also inherits from
ValueErrorfor compatibility
When Raised¶
Cache entry format is invalid
Data types don’t match schema
Required fields are missing
Validation constraints violated
Example¶
from proxywhirl.cache import CacheManager
from proxywhirl.exceptions import CacheValidationError
cache = CacheManager()
try:
# Invalid data structure
cache.set("proxy_stats", {"invalid": "format"})
stats = cache.get("proxy_stats")
except CacheValidationError as e:
print(f"Validation failed: {e}")
# Solution: Use correct data structure
cache.set("proxy_stats", {
"total_requests": 100,
"success_rate": 0.95,
"avg_latency_ms": 250.0
})
Resolution Steps¶
Verify cache entry format matches schema
Check data types are correct
Ensure required fields are present
Validate constraints (ranges, formats)
Clear invalid cache entries
RequestQueueFullError¶
Raised when the request queue is full and cannot accept more requests.
Error Code:
QUEUE_FULLRetry Recommended: Yes (wait for queue to drain)
When Raised¶
Request queue has reached maximum capacity
Too many concurrent requests being processed
Request rate exceeds processing capacity
Queue size configuration is too small for workload
Attributes¶
Same as ProxyWhirlError, plus:
queue_size(int | None): Maximum queue size (included in metadata)
Example¶
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import RequestQueueFullError
import time
# Configure with small queue for testing
rotator = ProxyWhirl(queue_size=10)
try:
# Flood the queue with requests
for i in range(100):
rotator.request("GET", f"https://httpbin.org/delay/{i}")
except RequestQueueFullError as e:
print(f"Queue full: {e}") # "Request queue is full (max size: 10)..."
# Solution 1: Wait for queue to drain
time.sleep(5)
rotator.request("GET", "https://httpbin.org/ip")
# Solution 2: Increase queue size
rotator = ProxyWhirl(queue_size=100)
# Solution 3: Implement request throttling
import asyncio
async def throttled_requests(urls, max_concurrent=10):
semaphore = asyncio.Semaphore(max_concurrent)
async def fetch(url):
async with semaphore:
return await rotator.request("GET", url)
tasks = [fetch(url) for url in urls]
return await asyncio.gather(*tasks)
Resolution Steps¶
Wait for pending requests to complete
Increase
queue_sizein configurationReduce request rate to match processing capacity
Implement request throttling or batching
Add more proxy workers to increase throughput
Monitor queue metrics to optimize size
RetryableError¶
Raised to signal that an operation should be retried by the retry executor.
Module:
proxywhirl.retryInherits:
Exception(notProxyWhirlError)
When Raised¶
Transient network failures during proxied requests
Temporary proxy unavailability
Retryable HTTP status codes (502, 503, 504)
Example¶
from proxywhirl import RetryableError
try:
response = make_request_through_proxy()
except RetryableError:
# RetryExecutor catches this and retries automatically
pass
NonRetryableError¶
Raised to signal that an operation should NOT be retried.
Module:
proxywhirl.retryInherits:
Exception(notProxyWhirlError)
When Raised¶
Authentication failures (401/407)
Invalid request format
Permanent proxy configuration errors
Example¶
from proxywhirl import NonRetryableError
try:
response = make_request_through_proxy()
except NonRetryableError:
# Do not retry - fix the underlying issue
logger.error("Non-retryable error, check proxy credentials")
RegexTimeoutError¶
Raised when regex compilation or matching exceeds the configured timeout.
Module:
proxywhirl.safe_regexInherits:
Exception(notProxyWhirlError)Purpose: ReDoS (Regular Expression Denial of Service) protection
When Raised¶
Regex pattern takes too long to compile
Regex matching exceeds timeout threshold
Catastrophic backtracking detected
Example¶
from proxywhirl import RegexTimeoutError
from proxywhirl.safe_regex import safe_match
try:
result = safe_match(pattern, text, timeout=1.0)
except RegexTimeoutError:
logger.warning("Regex timed out - possible ReDoS pattern")
RegexComplexityError¶
Raised when a regex pattern is too complex or potentially dangerous.
Module:
proxywhirl.safe_regexInherits:
Exception(notProxyWhirlError)Purpose: Prevent ReDoS attacks from user-provided patterns
When Raised¶
Pattern contains nested quantifiers (e.g.,
(a+)+)Pattern exceeds complexity threshold
Pattern contains known ReDoS-vulnerable constructs
Example¶
from proxywhirl import RegexComplexityError
from proxywhirl.safe_regex import safe_compile
try:
pattern = safe_compile(user_provided_pattern)
except RegexComplexityError:
logger.warning("Regex pattern rejected - too complex")
Error Handling Patterns¶
Basic Error Handling¶
Catch specific exceptions for targeted error handling:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import (
ProxyPoolEmptyError,
ProxyConnectionError,
ProxyAuthenticationError,
)
rotator = ProxyWhirl()
try:
response = rotator.request("GET", "https://httpbin.org/ip")
print(f"Success! IP: {response.json()['origin']}")
except ProxyPoolEmptyError:
print("No proxies available - adding fallback proxy")
rotator.add_proxy("http://fallback-proxy.example.com:8080")
except ProxyAuthenticationError as e:
print(f"Authentication failed: {e}")
# Log error and alert admin
logger.error("Proxy credentials invalid", extra=e.to_dict())
except ProxyConnectionError as e:
print(f"Connection failed: {e}")
if e.retry_recommended:
# Retry with longer timeout
response = rotator.request("GET", url, timeout=60)
Catch-All Error Handling¶
Use ProxyWhirlError to catch all library-specific errors:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyWhirlError
rotator = ProxyWhirl()
try:
response = rotator.request("GET", "https://httpbin.org/ip")
except ProxyWhirlError as e:
# Structured error logging
print(f"ProxyWhirl error: {e}")
logger.error("Request failed", extra={
"error_code": e.error_code.value,
"proxy_url": e.proxy_url,
"retry_recommended": e.retry_recommended,
"attempt_count": e.attempt_count,
})
Retry Logic Based on Error Type¶
Implement smart retry logic based on error metadata:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyWhirlError
import time
def request_with_smart_retry(rotator, url, max_attempts=3):
"""Make request with smart retry logic."""
for attempt in range(max_attempts):
try:
return rotator.request("GET", url)
except ProxyWhirlError as e:
if not e.retry_recommended:
# Don't retry non-retryable errors
raise
if attempt < max_attempts - 1:
# Exponential backoff
delay = 2 ** attempt
print(f"Attempt {attempt + 1} failed, retrying in {delay}s...")
time.sleep(delay)
else:
# Final attempt failed
raise
Async Error Handling¶
Handle errors in async contexts:
import asyncio
from proxywhirl import AsyncProxyWhirl
from proxywhirl.exceptions import (
ProxyPoolEmptyError,
ProxyConnectionError,
)
async def fetch_with_fallback(url):
"""Fetch URL with automatic fallback strategy."""
rotator = AsyncProxyWhirl()
try:
response = await rotator.request("GET", url)
return response.json()
except ProxyPoolEmptyError:
# Fallback: Auto-fetch proxies
print("No proxies available, fetching from sources...")
await rotator.auto_fetch()
response = await rotator.request("GET", url)
return response.json()
except ProxyConnectionError as e:
# Fallback: Try without proxy
print(f"All proxies failed ({e}), trying direct connection...")
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.json()
# Usage
result = asyncio.run(fetch_with_fallback("https://httpbin.org/ip"))
REST API Error Handling¶
See also
For the full REST API error code reference, see REST API.
Handle exceptions in REST API endpoints:
from fastapi import FastAPI, HTTPException, status
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import (
ProxyWhirlError,
ProxyPoolEmptyError,
ProxyConnectionError,
)
app = FastAPI()
rotator = ProxyWhirl()
@app.post("/api/v1/request")
async def proxied_request(url: str):
"""Make proxied request with proper error handling."""
try:
response = rotator.request("GET", url)
return {
"status": "success",
"data": response.json(),
"proxy_used": response.headers.get("X-Proxy-ID")
}
except ProxyPoolEmptyError as e:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={
"error_code": e.error_code.value,
"message": str(e),
"suggestion": "Add proxies to the pool"
}
)
except ProxyConnectionError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"error_code": e.error_code.value,
"message": str(e),
"retry_recommended": e.retry_recommended
}
)
except ProxyWhirlError as e:
# Generic error handler for all other ProxyWhirl errors
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=e.to_dict()
)
Context Manager Pattern¶
Use context managers for automatic cleanup on errors:
from contextlib import contextmanager
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyWhirlError
@contextmanager
def rotator_session(storage_path=None):
"""Context manager for ProxyWhirl with automatic cleanup."""
rotator = ProxyWhirl(storage_path=storage_path)
try:
yield rotator
except ProxyWhirlError as e:
# Log error
print(f"Session error: {e}")
raise
finally:
# Cleanup (close connections, save state, etc.)
rotator.close()
# Usage
with rotator_session("/tmp/proxies.db") as rotator:
rotator.add_proxy("http://proxy.example.com:8080")
response = rotator.request("GET", "https://httpbin.org/ip")
# Automatic cleanup on exit
Error Aggregation¶
Aggregate multiple errors for batch operations:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyValidationError
from typing import List, Tuple
def add_proxies_bulk(
rotator: ProxyWhirl,
proxy_urls: List[str]
) -> Tuple[int, List[Tuple[str, Exception]]]:
"""Add multiple proxies, collecting errors."""
success_count = 0
errors = []
for url in proxy_urls:
try:
rotator.add_proxy(url)
success_count += 1
except ProxyValidationError as e:
errors.append((url, e))
return success_count, errors
# Usage
urls = [
"http://proxy1.example.com:8080",
"invalid-url", # Will fail
"http://proxy2.example.com:8080",
]
rotator = ProxyWhirl()
success, errors = add_proxies_bulk(rotator, urls)
print(f"Added {success} proxies")
for url, error in errors:
print(f"Failed to add {url}: {error}")
Best Practices¶
1. Use Specific Exceptions¶
Catch specific exceptions rather than generic ones:
# Good: Specific exception handling
try:
response = rotator.request("GET", url)
except ProxyPoolEmptyError:
rotator.auto_fetch()
except ProxyAuthenticationError:
update_credentials()
# Bad: Too broad
try:
response = rotator.request("GET", url)
except Exception:
# Can't distinguish between different errors
pass
2. Check retry_recommended Flag¶
Respect the retry_recommended flag to avoid retrying non-retryable errors:
from proxywhirl.exceptions import ProxyWhirlError
try:
response = rotator.request("GET", url)
except ProxyWhirlError as e:
if e.retry_recommended:
# Retry makes sense
time.sleep(1)
response = rotator.request("GET", url)
else:
# Don't retry - error requires manual intervention
logger.error(f"Non-retryable error: {e}")
raise
3. Use Error Codes for Programmatic Handling¶
Use error codes for reliable programmatic error handling:
from proxywhirl.exceptions import ProxyWhirlError, ProxyErrorCode
try:
response = rotator.request("GET", url)
except ProxyWhirlError as e:
# More reliable than string matching
if e.error_code == ProxyErrorCode.PROXY_POOL_EMPTY:
handle_empty_pool()
elif e.error_code == ProxyErrorCode.TIMEOUT:
handle_timeout()
4. Log Structured Error Data¶
Use to_dict() for structured logging:
from proxywhirl.exceptions import ProxyWhirlError
import json
try:
response = rotator.request("GET", url)
except ProxyWhirlError as e:
# Structured logging with all error metadata
logger.error(
"Request failed",
extra=e.to_dict()
)
# Or for JSON logging
print(json.dumps(e.to_dict(), indent=2))
5. Implement Circuit Breaker Pattern¶
Use circuit breakers to prevent cascading failures:
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import ProxyConnectionError
rotator = ProxyWhirl(
circuit_breaker_enabled=True,
failure_threshold=5,
recovery_timeout=60
)
try:
response = rotator.request("GET", url)
except ProxyConnectionError as e:
if "circuit breaker is open" in str(e).lower():
# Circuit breaker is protecting against failing proxy
# Wait for recovery or remove proxy
print("Circuit breaker open, waiting for recovery...")
time.sleep(60)
6. Provide User-Friendly Error Messages¶
Catch ProxyWhirl errors and provide context-specific messages:
from proxywhirl.exceptions import (
ProxyPoolEmptyError,
ProxyConnectionError,
ProxyAuthenticationError,
)
def user_friendly_request(url):
"""Make request with user-friendly error messages."""
try:
return rotator.request("GET", url)
except ProxyPoolEmptyError:
return {
"success": False,
"error": "No proxies configured. Please add proxies first.",
"action": "add_proxies"
}
except ProxyAuthenticationError:
return {
"success": False,
"error": "Proxy credentials are invalid. Please check your username and password.",
"action": "update_credentials"
}
except ProxyConnectionError as e:
return {
"success": False,
"error": f"Unable to connect through proxy: {e}",
"action": "check_proxy_status"
}
7. Test Error Handling¶
Write tests for error scenarios:
import pytest
from proxywhirl import ProxyWhirl
from proxywhirl.exceptions import (
ProxyPoolEmptyError,
ProxyValidationError,
)
def test_empty_pool_error():
"""Test that empty pool raises appropriate error."""
rotator = ProxyWhirl()
with pytest.raises(ProxyPoolEmptyError) as exc_info:
rotator.request("GET", "https://httpbin.org/ip")
assert exc_info.value.error_code.value == "PROXY_POOL_EMPTY"
assert exc_info.value.retry_recommended is False
def test_invalid_proxy_error():
"""Test that invalid proxy URL raises validation error."""
rotator = ProxyWhirl()
with pytest.raises(ProxyValidationError) as exc_info:
rotator.add_proxy("invalid-url")
assert exc_info.value.error_code.value == "PROXY_VALIDATION_FAILED"
8. Monitor Error Rates¶
Track error rates to identify proxy quality issues:
from collections import defaultdict
from proxywhirl.exceptions import ProxyWhirlError
error_counts = defaultdict(int)
def monitored_request(url):
"""Make request with error monitoring."""
try:
return rotator.request("GET", url)
except ProxyWhirlError as e:
# Track error types
error_counts[e.error_code.value] += 1
# Alert if error rate is high
total_errors = sum(error_counts.values())
if total_errors > 100:
print(f"High error rate detected: {error_counts}")
raise
# Periodic reporting
def report_error_stats():
"""Report error statistics."""
print("Error Statistics:")
for error_code, count in error_counts.items():
print(f" {error_code}: {count}")
9. Handle Async Errors Properly¶
Use proper async error handling patterns:
import asyncio
from proxywhirl import AsyncProxyWhirl
from proxywhirl.exceptions import ProxyWhirlError
async def async_request_with_timeout(url, timeout=30):
"""Async request with timeout and error handling."""
rotator = AsyncProxyWhirl()
try:
# Add timeout to prevent hanging
response = await asyncio.wait_for(
rotator.request("GET", url),
timeout=timeout
)
return response
except asyncio.TimeoutError:
print(f"Request timed out after {timeout}s")
raise
except ProxyWhirlError as e:
print(f"ProxyWhirl error: {e}")
raise
finally:
await rotator.close()
10. Document Error Handling¶
Document expected errors in docstrings:
def fetch_data(url: str) -> dict:
"""
Fetch data from URL through proxy.
Args:
url: Target URL to fetch
Returns:
Parsed JSON response
Raises:
ProxyPoolEmptyError: No proxies available in pool
ProxyConnectionError: Unable to connect through proxy
ProxyAuthenticationError: Proxy authentication failed
Example:
>>> try:
... data = fetch_data("https://api.example.com/data")
... except ProxyPoolEmptyError:
... rotator.auto_fetch()
... data = fetch_data("https://api.example.com/data")
"""
response = rotator.request("GET", url)
return response.json()
Summary¶
ProxyWhirl’s exception system provides:
Hierarchical structure - All exceptions inherit from
ProxyWhirlErrorError codes - Programmatic error handling with
ProxyErrorCodeRich metadata - Proxy URL, attempt count, retry recommendations
Security - Automatic URL redaction to protect credentials
Actionable guidance - Each exception includes resolution steps
Use specific exception types for targeted error handling, check the retry_recommended flag, and leverage error codes for reliable programmatic handling.
See Also¶
Python API – Main ProxyWhirl API reference
REST API – REST API error codes and responses
Cache API – Cache-specific error handling
Retry & Failover – Retry patterns and circuit breaker integration
Async Client – Async error handling patterns
Deployment Security – Production error monitoring
For questions or issues: