Error Handling

Request Forge provides a comprehensive exception hierarchy for robust error handling.

Exception Hierarchy

All HTTP Client exceptions inherit from HttpClientException:

HttpClientException (base)
├── MaxRetryException          # Maximum retries exceeded
├── TimeoutException           # Request timeout
├── ConnectionException        # Network/connection errors
├── SSLException               # SSL/TLS errors
├── ResponseParseException     # JSON/response parsing errors
├── AuthenticationException    # Authentication failures
│   └── TokenRefreshException  # Token refresh failures
└── HttpStatusException        # HTTP error status codes
    ├── BadRequestException      # 400 Bad Request
    ├── UnauthorizedException    # 401 Unauthorized
    ├── ForbiddenException       # 403 Forbidden
    ├── NotFoundException        # 404 Not Found
    └── ServerErrorException     # 5xx Server Errors

Base Exception

HttpClientException

All exceptions inherit from this base class:

from requestforge import HttpClientException

try:
    response = client.get('/users/1')
except HttpClientException as e:
    print(f"Error: {e.message}")
    print(f"Original exception: {e.original_exception}")
    print(f"Context: {e.context}")

Attributes:

  • message (str): Human-readable error message

  • original_exception (Exception | None): Original underlying exception

  • context (dict): Additional context about the error

Network Errors

TimeoutException

Raised when a request times out:

from requestforge import TimeoutException

try:
    response = client.get('/slow-endpoint', timeout=5.0)
except TimeoutException as e:
    print(f"Request timed out after 5 seconds")
    print(f"URL: {e.context.get('request').url}")

When raised:

  • Request exceeds configured timeout

  • Connection timeout

  • Read timeout

Best practices:

# Handle timeout with retry
from requestforge import TimeoutException
import time

max_attempts = 3
for attempt in range(max_attempts):
    try:
        response = client.get('/data', timeout=10.0)
        break
    except TimeoutException:
        if attempt < max_attempts - 1:
            time.sleep(2 ** attempt)  # Exponential backoff
            continue
        raise

ConnectionException

Raised on network connection failures:

from requestforge import ConnectionException

try:
    response = client.get('/users')
except ConnectionException as e:
    print(f"Network error: {e.message}")
    # Check if server is unreachable
    if "Connection refused" in str(e.original_exception):
        print("Server is not running")

Common causes:

  • Server is down or unreachable

  • DNS resolution failure

  • Network connectivity issues

  • Firewall blocking connection

Handling:

from requestforge import ConnectionException
import logging

try:
    response = client.get('/api/data')
except ConnectionException as e:
    logging.error(f"Cannot connect to API: {e}")
    # Fallback to cached data or show error to user
    return get_cached_data()

SSLException

Raised on SSL/TLS errors:

from requestforge import SSLException

try:
    response = client.get('https://self-signed.badssl.com')
except SSLException as e:
    print(f"SSL error: {e.message}")
    print(f"Original: {e.original_exception}")

Common causes:

  • Invalid SSL certificate

  • Self-signed certificate

  • Certificate expired

  • Hostname mismatch

Disabling SSL verification (not recommended for production):

config = (
    HttpClientConfigBuilder()
    .with_base_url('https://api.example.com')
    .with_verify_ssl(False)  # ⚠️ Security risk!
    .build()
)

HTTP Status Errors

HttpStatusException

Base class for HTTP status code errors:

from requestforge import HttpStatusException

try:
    response = client.get('/users/999')
except HttpStatusException as e:
    print(f"HTTP {e.status_code}: {e.message}")
    print(f"Response body: {e.response_body}")

Attributes:

  • status_code (int): HTTP status code

  • response_body (str | None): Response body content

  • message (str): Error message

  • original_exception (Exception | None): Original exception

  • context (dict): Request context

BadRequestException (400)

Raised for 400 Bad Request responses:

from requestforge import BadRequestException

try:
    response = client.post('/users', json_data={
        'email': 'invalid-email'  # Invalid data
    })
except BadRequestException as e:
    print(f"Invalid request: {e.response_body}")
    # Parse error details
    error_data = json.loads(e.response_body)
    for field, errors in error_data.get('errors', {}).items():
        print(f"{field}: {errors}")

UnauthorizedException (401)

Raised for 401 Unauthorized responses:

from requestforge import UnauthorizedException

try:
    response = client.get('/protected-resource')
except UnauthorizedException:
    print("Authentication required or token expired")
    # Redirect to login or refresh token

Note: When using TokenAuthHook, 401 errors trigger automatic token refresh and retry.

ForbiddenException (403)

Raised for 403 Forbidden responses:

from requestforge import ForbiddenException

try:
    response = client.delete('/admin/users/1')
except ForbiddenException:
    print("Insufficient permissions")
    # Show error to user or request elevated access

NotFoundException (404)

Raised for 404 Not Found responses:

from requestforge import NotFoundException

try:
    response = client.get('/users/nonexistent')
except NotFoundException:
    print("Resource not found")
    return None  # Or show 404 page

ServerErrorException (5xx)

Raised for 5xx server errors:

from requestforge import ServerErrorException

try:
    response = client.get('/api/data')
except ServerErrorException as e:
    print(f"Server error {e.status_code}")
    if e.status_code == 503:
        print("Service temporarily unavailable")
    elif e.status_code == 500:
        print("Internal server error")

Retry Errors

MaxRetryException

Raised when maximum retry attempts are exceeded:

from requestforge import MaxRetryException

try:
    response = client.get('/unstable-endpoint')
except MaxRetryException as e:
    print(f"Failed after {e.attempts} attempts")
    print(f"Last error: {e.original_exception}")
    print(f"Request: {e.context.get('request')}")

Attributes:

  • attempts (int): Total number of attempts made

  • original_exception (Exception | None): Last exception that caused failure

  • message (str): Error message

  • context (dict): Request context

Example with custom handling:

from requestforge import MaxRetryException, TimeoutException

try:
    response = client.get('/api/data')
except MaxRetryException as e:
    if isinstance(e.original_exception, TimeoutException):
        print("Service is responding too slowly")
        # Maybe increase timeout or reduce load
    else:
        print(f"Service failed: {e.original_exception}")
        # Log to monitoring system

Authentication Errors

AuthenticationException

Raised on authentication failures:

from requestforge import AuthenticationException

try:
    response = client.get('/protected')
except AuthenticationException as e:
    print(f"Auth failed for service: {e.service_name}")
    print(f"Reason: {e.message}")
    # Clear cached credentials or redirect to login

Attributes:

  • service_name (str | None): Name of the service that failed

  • message (str): Error message

  • original_exception (Exception | None): Original exception

  • context (dict): Additional context

TokenRefreshException

Raised when token refresh fails:

from requestforge.token_manager import TokenRefreshException

try:
    token = token_manager.get_token()
except TokenRefreshException as e:
    print(f"Failed to refresh token: {e}")
    # Clear session and redirect to login

Parsing Errors

ResponseParseException

Raised when response parsing fails:

from requestforge import ResponseParseException

try:
    response = client.get('/data')
    data = response.json()  # Expects JSON
except ResponseParseException as e:
    print(f"Invalid JSON response: {e.message}")
    print(f"Content preview: {e.context.get('content_preview')}")
    # Use response.text instead or handle as plain text

Common causes:

  • Server returns HTML instead of JSON

  • Malformed JSON

  • Empty response when JSON expected

Safe parsing:

response = client.get('/data')

# Option 1: Use json_or_none()
data = response.json_or_none()
if data is None:
    print("Response is not JSON")

# Option 2: Try-except
try:
    data = response.json()
except ResponseParseException:
    # Fallback to text
    data = response.text

Error Handling Patterns

Specific to General

Handle specific exceptions first, then general:

from requestforge import (
    NotFoundException,
    UnauthorizedException,
    HttpStatusException,
    TimeoutException,
    HttpClientException
)

try:
    response = client.get('/users/1')
    user = response.json()

# Most specific first
except NotFoundException:
    print("User not found")
    return None

except UnauthorizedException:
    print("Please login")
    redirect_to_login()

# More general HTTP errors
except HttpStatusException as e:
    print(f"HTTP error {e.status_code}")

# Network errors
except TimeoutException:
    print("Request timed out")

# Catch-all for HTTP client errors
except HttpClientException as e:
    print(f"Request failed: {e}")
    logger.error(f"Error details: {e.context}")

Retry Pattern

Implement custom retry with specific error handling:

from requestforge import TimeoutException, ServerErrorException
import time

def fetch_with_retry(url, max_retries=3):
    last_error = None

    for attempt in range(max_retries):
        try:
            response = client.get(url)
            return response.json()

        except TimeoutException as e:
            print(f"Timeout on attempt {attempt + 1}")
            last_error = e
            time.sleep(2 ** attempt)  # Exponential backoff

        except ServerErrorException as e:
            if e.status_code == 503:  # Service Unavailable
                print(f"Service unavailable, attempt {attempt + 1}")
                last_error = e
                time.sleep(5)
            else:
                raise  # Don't retry other server errors

        except HttpClientException as e:
            # Don't retry other errors
            raise

    # Max retries exceeded
    raise MaxRetryException(
        message=f"Failed after {max_retries} attempts",
        attempts=max_retries,
        original_exception=last_error
    )

Fallback Pattern

Provide fallback data on errors:

from requestforge import HttpClientException

def get_user_data(user_id):
    try:
        response = client.get(f'/users/{user_id}')
        return response.json()

    except NotFoundException:
        return None

    except HttpClientException as e:
        logger.error(f"Failed to fetch user {user_id}: {e}")
        # Return cached data if available
        return get_cached_user_data(user_id)

Circuit Breaker Pattern

Stop making requests after consecutive failures:

from requestforge import HttpClientException
import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half_open

    def call(self, func):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'half_open'
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func()
            self.on_success()
            return result
        except HttpClientException as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        self.state = 'closed'

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = 'open'

# Usage
breaker = CircuitBreaker()

try:
    response = breaker.call(lambda: client.get('/api/data'))
except Exception as e:
    print(f"Request failed or circuit open: {e}")

Context Managers

Ensure cleanup even on errors:

from requestforge import http_client, HttpClientException

try:
    with http_client('https://api.example.com') as client:
        response = client.get('/users')
        users = response.json()
        # Process users
except HttpClientException as e:
    print(f"API error: {e}")
# Client automatically closed

Logging Errors

Basic Logging

import logging
from requestforge import HttpClientException

logger = logging.getLogger(__name__)

try:
    response = client.get('/api/data')
except HttpClientException as e:
    logger.error(
        f"API request failed: {e.message}",
        exc_info=True,  # Include stack trace
        extra={
            'url': e.context.get('request', {}).get('url'),
            'exception_type': type(e).__name__
        }
    )

Structured Logging

import structlog
from requestforge import HttpClientException, HttpStatusException

logger = structlog.get_logger()

try:
    response = client.get('/users/1')
except HttpStatusException as e:
    logger.error(
        "http_request_failed",
        status_code=e.status_code,
        url=e.context.get('request', {}).get('url'),
        response_body=e.response_body[:200],
        exception_type=type(e).__name__
    )
except HttpClientException as e:
    logger.error(
        "http_request_error",
        error=str(e),
        exception_type=type(e).__name__,
        context=e.context
    )

Error Hooks

Use error hooks for centralized logging:

from requestforge.interfaces import ErrorHookInterface
import logging

class ErrorLoggingHook(ErrorHookInterface):
    def __init__(self):
        self.logger = logging.getLogger('requestforge')

    def on_error(self, exception, context):
        self.logger.error(
            f"Request failed: {type(exception).__name__}",
            extra={
                'exception': str(exception),
                'url': context.request.url,
                'method': context.request.method.value,
                'attempt': context.attempt,
                'max_retries': context.max_retries
            }
        )

config = builder.with_error_hook(ErrorLoggingHook()).build()

Error Monitoring

Sentry Integration

import sentry_sdk
from requestforge import HttpClientException
from requestforge.interfaces import ErrorHookInterface

class SentryErrorHook(ErrorHookInterface):
    def on_error(self, exception, context):
        with sentry_sdk.push_scope() as scope:
            # Add tags
            scope.set_tag("http_method", context.request.method.value)
            scope.set_tag("http_url", context.request.url)

            # Add context
            scope.set_context("request", {
                "url": context.request.url,
                "method": context.request.method.value,
                "attempt": context.attempt,
                "max_retries": context.max_retries
            })

            # Capture exception
            sentry_sdk.capture_exception(exception)

config = builder.with_error_hook(SentryErrorHook()).build()

Datadog Integration

from datadog import statsd
from requestforge.interfaces import ErrorHookInterface

class DatadogErrorHook(ErrorHookInterface):
    def on_error(self, exception, context):
        # Increment error counter
        statsd.increment(
            'requestforge.error',
            tags=[
                f'exception_type:{type(exception).__name__}',
                f'url:{context.request.url}',
                f'method:{context.request.method.value}'
            ]
        )

Best Practices

  1. Handle Specific Exceptions

    # Good ✅ - Specific handling
try:
    response = client.get('/users/1')
except NotFoundException:
    return None
except UnauthorizedException:
    redirect_to_login()

# Avoid ❌ - Too broad
try:
    response = client.get('/users/1')
except Exception:
    print("Something went wrong")

  2. Don’t Silence Errors

    # Good ✅ - Log and re-raise or handle
try:
    response = client.get('/data')
except HttpClientException as e:
    logger.error(f"Request failed: {e}")
    raise

# Avoid ❌ - Silent failure
try:
    response = client.get('/data')
except:
    pass

  3. Provide Context

    # Good ✅ - Rich error information
try:
    response = client.get(f'/users/{user_id}')
except HttpClientException as e:
    raise ValueError(
        f"Failed to fetch user {user_id}: {e.message}"
    ) from e

  4. Use Appropriate Retry

    # Good ✅ - Retry on transient errors
config = builder.with_retry(max_retries=3).build()

# Transient errors automatically retried:
# - TimeoutException
# - ConnectionException
# - 5xx ServerErrorException

  5. Clean Up Resources

    # Good ✅ - Context manager ensures cleanup
with http_client('https://api.example.com') as client:
    response = client.get('/data')

# Or explicit cleanup
client = create_client('https://api.example.com')
try:
    response = client.get('/data')
finally:
    client.close()

Testing Error Handling

Mock Exceptions

import pytest
from unittest.mock import patch, Mock
from requestforge import TimeoutException, NotFoundException

def test_timeout_handling():
    with patch('requestforge.HttpClient.get') as mock_get:
        mock_get.side_effect = TimeoutException("Timeout")

        # Your code that handles timeout
        result = fetch_user_with_timeout_handling(1)

        assert result is None

def test_not_found_handling():
    with patch('requestforge.HttpClient.get') as mock_get:
        mock_get.side_effect = NotFoundException("Not found")

        result = get_user(999)

        assert result is None

Mock HTTP Responses

import responses
from requestforge import HttpClient, NotFoundException

@responses.activate
def test_404_handling():
    responses.add(
        responses.GET,
        'https://api.example.com/users/999',
        status=404,
        json={'error': 'User not found'}
    )

    client = create_client('https://api.example.com')

    with pytest.raises(NotFoundException):
        client.get('/users/999')

Common Scenarios

API Rate Limiting

from requestforge import HttpStatusException
import time

def api_call_with_rate_limit(url):
    while True:
        try:
            response = client.get(url)
            return response.json()

        except HttpStatusException as e:
            if e.status_code == 429:  # Too Many Requests
                retry_after = int(e.response.headers.get('Retry-After', 60))
                print(f"Rate limited, waiting {retry_after}s")
                time.sleep(retry_after)
            else:
                raise

Partial Failures

from requestforge import HttpClientException

def fetch_multiple_users(user_ids):
    results = []
    errors = []

    for user_id in user_ids:
        try:
            response = client.get(f'/users/{user_id}')
            results.append(response.json())
        except HttpClientException as e:
            errors.append({
                'user_id': user_id,
                'error': str(e)
            })

    return results, errors

Graceful Degradation

from requestforge import HttpClientException

def get_recommendations(user_id):
    try:
        # Try personalized recommendations
        response = client.get(f'/recommendations/{user_id}')
        return response.json()

    except HttpClientException:
        # Fall back to popular items
        try:
            response = client.get('/recommendations/popular')
            return response.json()
        except HttpClientException:
            # Final fallback to cached data
            return get_default_recommendations()

Next Steps