🚨 Error Handling

Copy

• Copy for LLM

  Copy page as Markdown for LLMs
• View as Markdown

  Open this page as Markdown
• Open in ChatGPT

  Get insights from ChatGPT
• Open in Claude

  Get insights from Claude

{
  "detail": "Human-readable error description",
  "status_code": 400,
  "error_type": "ValidationError",
  "meta": {
    "timestamp": "2024-01-01T00:00:00Z",
    "request_id": "req_123456789"
  },
  "errors": [
    {
      "field": "email",
      "message": "Invalid email format",
      "code": "INVALID_FORMAT"
    }
  ]
}

{
  "detail": "Invalid API key",
  "status_code": 401,
  "error_type": "AuthenticationError"
}

def handle_auth_error(response):
    if response.status_code == 401:
        error_data = response.json()
        if "Invalid API key" in error_data.get("detail", ""):
            # Check API key configuration
            api_key = os.getenv('FREDDY_API_KEY')
            if not api_key:
                raise ValueError("FREDDY_API_KEY environment variable not set")
            if not api_key.startswith('ak_'):
                raise ValueError("Invalid API key format")
            # Key might be expired - generate new one
            print("API key may be expired. Please generate a new key.")

{
  "detail": "Validation failed",
  "status_code": 422,
  "error_type": "ValidationError",
  "errors": [
    {
      "field": "email",
      "message": "Invalid email format",
      "code": "INVALID_FORMAT"
    },
    {
      "field": "password",
      "message": "Password must be at least 8 characters",
      "code": "TOO_SHORT"
    }
  ]
}

def handle_validation_errors(response):
    if response.status_code in [400, 422]:
        error_data = response.json()
        if error_data.get("error_type") == "ValidationError":
            errors = error_data.get("errors", [])
            for error in errors:
                field = error.get("field")
                message = error.get("message")
                code = error.get("code")
                print(f"Validation error in {field}: {message} ({code})")
            return errors
    return None

{
  "detail": "Rate limit exceeded",
  "status_code": 429,
  "error_type": "RateLimitError",
  "retry_after": 60
}

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1640995200
Retry-After: 60

import time
from datetime import datetime

def handle_rate_limit(response):
    if response.status_code == 429:
        # Get retry delay from header or response body
        retry_after = response.headers.get('Retry-After')
        if not retry_after:
            error_data = response.json()
            retry_after = error_data.get('retry_after', 60)
        
        retry_after = int(retry_after)
        print(f"Rate limited. Waiting {retry_after} seconds...")
        time.sleep(retry_after)
        return True
    return False

{
  "detail": "Insufficient permissions to access this resource",
  "status_code": 403,
  "error_type": "PermissionError"
}

{
  "detail": "Thread not found",
  "status_code": 404,
  "error_type": "NotFoundError"
}

{
  "detail": "Email already exists",
  "status_code": 409,
  "error_type": "ConflictError"
}

import time
import random
from typing import Callable, Any

def exponential_backoff_retry(
    func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff_factor: float = 2.0,
    jitter: bool = True
) -> Any:
    """
    Retry function with exponential backoff.
    
    Args:
        func: Function to retry
        max_retries: Maximum number of retry attempts
        base_delay: Initial delay in seconds
        max_delay: Maximum delay in seconds
        backoff_factor: Multiplier for delay after each retry
        jitter: Add random jitter to prevent thundering herd
    """
    
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise e
            
            # Calculate delay
            delay = min(base_delay * (backoff_factor ** attempt), max_delay)
            
            # Add jitter
            if jitter:
                delay += random.uniform(0, delay * 0.1)
            
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
            time.sleep(delay)

# Usage
def make_api_request():
    response = requests.get(
        "https://api.freddy.ai/v2/models",
        headers={"api-key": os.getenv('FREDDY_API_KEY')}
    )
    response.raise_for_status()
    return response.json()

result = exponential_backoff_retry(make_api_request)

class SmartRetryClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"api-key": api_key}
    
    def should_retry(self, response) -> bool:
        """Determine if request should be retried."""
        # Always retry on server errors
        if response.status_code >= 500:
            return True
        
        # Retry on rate limits
        if response.status_code == 429:
            return True
        
        # Don't retry on client errors (except rate limits)
        if 400 <= response.status_code < 500:
            return False
        
        return False
    
    def get_retry_delay(self, response, attempt: int) -> float:
        """Calculate retry delay based on response."""
        if response.status_code == 429:
            # Use Retry-After header if available
            retry_after = response.headers.get('Retry-After')
            if retry_after:
                return float(retry_after)
        
        # Exponential backoff for server errors
        return min(2 ** attempt, 60)  # Cap at 60 seconds
    
    def make_request(self, method: str, endpoint: str, **kwargs) -> dict:
        """Make API request with smart retry logic."""
        url = f"https://api.freddy.ai/v2/{endpoint}"
        max_retries = 3
        
        for attempt in range(max_retries + 1):
            try:
                response = requests.request(
                    method, url, headers=self.headers, **kwargs
                )
                
                # Success
                if response.status_code < 400:
                    return response.json()
                
                # Check if we should retry
                if attempt < max_retries and self.should_retry(response):
                    delay = self.get_retry_delay(response, attempt)
                    print(f"Request failed (attempt {attempt + 1}). Retrying in {delay}s...")
                    time.sleep(delay)
                    continue
                
                # No more retries or shouldn't retry
                response.raise_for_status()
                
            except requests.exceptions.RequestException as e:
                if attempt < max_retries:
                    delay = 2 ** attempt
                    print(f"Network error (attempt {attempt + 1}): {e}. Retrying in {delay}s...")
                    time.sleep(delay)
                    continue
                raise
        
        raise Exception("Max retries exceeded")

# Usage
client = SmartRetryClient(os.getenv('FREDDY_API_KEY'))
result = client.make_request("GET", "models")

import logging
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Any

class ErrorSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class ErrorContext:
    endpoint: str
    method: str
    request_data: Optional[Dict[str, Any]] = None
    user_id: Optional[str] = None
    organization_id: Optional[str] = None

class FreddyErrorHandler:
    def __init__(self, logger: Optional[logging.Logger] = None):
        self.logger = logger or logging.getLogger(__name__)
    
    def handle_error(self, response, context: ErrorContext) -> Dict[str, Any]:
        """Comprehensive error handling."""
        try:
            error_data = response.json()
        except:
            error_data = {"detail": "Unknown error", "status_code": response.status_code}
        
        error_info = {
            "status_code": response.status_code,
            "error_type": error_data.get("error_type", "UnknownError"),
            "detail": error_data.get("detail", "Unknown error"),
            "request_id": error_data.get("meta", {}).get("request_id"),
            "context": context,
            "severity": self._determine_severity(response.status_code),
            "recoverable": self._is_recoverable(response.status_code),
            "suggested_action": self._get_suggested_action(response.status_code, error_data)
        }
        
        # Log the error
        self._log_error(error_info)
        
        # Handle specific error types
        if response.status_code == 401:
            return self._handle_auth_error(error_info)
        elif response.status_code == 403:
            return self._handle_permission_error(error_info)
        elif response.status_code == 429:
            return self._handle_rate_limit_error(error_info, response)
        elif response.status_code == 422:
            return self._handle_validation_error(error_info, error_data)
        elif response.status_code >= 500:
            return self._handle_server_error(error_info)
        else:
            return self._handle_client_error(error_info)
    
    def _determine_severity(self, status_code: int) -> ErrorSeverity:
        """Determine error severity based on status code."""
        if status_code >= 500:
            return ErrorSeverity.HIGH
        elif status_code == 429:
            return ErrorSeverity.MEDIUM
        elif status_code in [401, 403]:
            return ErrorSeverity.MEDIUM
        else:
            return ErrorSeverity.LOW
    
    def _is_recoverable(self, status_code: int) -> bool:
        """Determine if error is recoverable."""
        # Server errors and rate limits are recoverable
        if status_code >= 500 or status_code == 429:
            return True
        # Auth errors might be recoverable (refresh token, new API key)
        if status_code in [401, 403]:
            return True
        # Most client errors are not recoverable
        return False
    
    def _get_suggested_action(self, status_code: int, error_data: Dict) -> str:
        """Get suggested action for error."""
        if status_code == 401:
            return "Check API key validity and regenerate if necessary"
        elif status_code == 403:
            return "Verify account permissions and upgrade plan if needed"
        elif status_code == 429:
            return "Implement rate limiting and retry with exponential backoff"
        elif status_code == 422:
            return "Fix validation errors in request data"
        elif status_code == 404:
            return "Verify resource ID exists and is accessible"
        elif status_code >= 500:
            return "Retry request with exponential backoff"
        else:
            return "Review request format and parameters"
    
    def _log_error(self, error_info: Dict[str, Any]):
        """Log error with appropriate level."""
        severity = error_info["severity"]
        message = f"API Error: {error_info['detail']}"
        
        extra = {
            "status_code": error_info["status_code"],
            "error_type": error_info["error_type"],
            "request_id": error_info["request_id"],
            "endpoint": error_info["context"].endpoint,
            "method": error_info["context"].method,
            "recoverable": error_info["recoverable"]
        }
        
        if severity == ErrorSeverity.CRITICAL:
            self.logger.critical(message, extra=extra)
        elif severity == ErrorSeverity.HIGH:
            self.logger.error(message, extra=extra)
        elif severity == ErrorSeverity.MEDIUM:
            self.logger.warning(message, extra=extra)
        else:
            self.logger.info(message, extra=extra)
    
    def _handle_auth_error(self, error_info: Dict) -> Dict[str, Any]:
        """Handle authentication errors."""
        return {
            **error_info,
            "recovery_steps": [
                "Verify API key is set correctly",
                "Check API key format (should start with 'ak_')",
                "Generate new API key if current one is expired",
                "Ensure using correct environment (dev/prod)"
            ]
        }
    
    def _handle_rate_limit_error(self, error_info: Dict, response) -> Dict[str, Any]:
        """Handle rate limit errors."""
        retry_after = response.headers.get('Retry-After', '60')
        
        return {
            **error_info,
            "retry_after": int(retry_after),
            "recovery_steps": [
                f"Wait {retry_after} seconds before retrying",
                "Implement exponential backoff",
                "Consider upgrading to higher rate limit plan",
                "Optimize request patterns to reduce frequency"
            ]
        }
    
    def _handle_validation_error(self, error_info: Dict, error_data: Dict) -> Dict[str, Any]:
        """Handle validation errors."""
        validation_errors = error_data.get("errors", [])
        
        return {
            **error_info,
            "validation_errors": validation_errors,
            "recovery_steps": [
                "Fix validation errors in request data",
                "Check API documentation for required fields",
                "Validate data format and types",
                "Ensure all required parameters are provided"
            ]
        }
    
    def _handle_server_error(self, error_info: Dict) -> Dict[str, Any]:
        """Handle server errors."""
        return {
            **error_info,
            "recovery_steps": [
                "Retry request with exponential backoff",
                "Check API status page for service issues",
                "Contact support if error persists",
                "Implement circuit breaker pattern"
            ]
        }
    
    def _handle_client_error(self, error_info: Dict) -> Dict[str, Any]:
        """Handle other client errors."""
        return {
            **error_info,
            "recovery_steps": [
                "Review request format and parameters",
                "Check API documentation for endpoint requirements",
                "Verify resource IDs and permissions",
                "Ensure request data is properly formatted"
            ]
        }

# Usage
error_handler = FreddyErrorHandler()

def make_api_request_with_error_handling(endpoint, method="GET", **kwargs):
    context = ErrorContext(
        endpoint=endpoint,
        method=method,
        request_data=kwargs.get("json"),
        user_id="user_123"  # Get from session
    )
    
    try:
        response = requests.request(
            method,
            f"https://api.freddy.ai/v2/{endpoint}",
            headers={"api-key": os.getenv('FREDDY_API_KEY')},
            **kwargs
        )
        
        if response.status_code >= 400:
            error_info = error_handler.handle_error(response, context)
            
            # Handle based on error type
            if error_info["recoverable"] and error_info["status_code"] == 429:
                time.sleep(error_info.get("retry_after", 60))
                # Retry the request
                return make_api_request_with_error_handling(endpoint, method, **kwargs)
            
            raise Exception(f"API Error: {error_info['detail']}")
        
        return response.json()
        
    except requests.exceptions.RequestException as e:
        context_dict = {
            "endpoint": endpoint,
            "method": method,
            "error": str(e)
        }
        error_handler.logger.error("Network error", extra=context_dict)
        raise