🚨 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

{
  "success": false,
  "error": {
    "code": "INVALID_EMAIL_FORMAT",
    "message": "Please enter a valid email address.",
    "system_message": "Email validation failed",
    "type": "client_error",
    "status": 422,
    "details": {
      "field": "email",
      "value": "invalid-email"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

{
  "success": false,
  "error": {
    "code": "INVALID_CREDENTIALS",
    "message": "Invalid email, username, or password. Please try again.",
    "system_message": "Authentication failed",
    "type": "authentication_error",
    "status": 401,
    "details": {
      "email": "user@example.com"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

def handle_auth_error(response):
    if response.status_code == 401:
        error_data = response.json()
        error_code = error_data.get("error", {}).get("code")
        
        if error_code == "INVALID_API_KEY":
            # Check API key configuration
            api_key = os.getenv('FREDDY_API_KEY')
            if not api_key:
                raise ValueError("FREDDY_API_KEY environment variable not set")
            # Key might be expired - generate new one
            print("API key may be expired. Please generate a new key.")
        elif error_code == "TOKEN_EXPIRED":
            # Refresh the token
            print("Token expired. Please refresh your session.")

{
  "success": false,
  "error": {
    "code": "INVALID_EMAIL_FORMAT",
    "message": "Please enter a valid email address.",
    "system_message": "Email validation failed",
    "type": "validation_error",
    "status": 422,
    "details": {
      "field": "email",
      "value": "invalid-email"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

def handle_validation_errors(response):
    if response.status_code == 422:
        error_data = response.json()
        error = error_data.get("error", {})
        
        code = error.get("code")
        message = error.get("message")
        details = error.get("details", {})
        
        print(f"Validation error ({code}): {message}")
        if "field" in details:
            print(f"Field: {details['field']}")
        
        return error
    return None

{
  "success": false,
  "error": {
    "code": "RATE_LIMIT_EXCEEDED",
    "message": "Too many requests. Please wait a moment and try again.",
    "system_message": "Rate limit exceeded",
    "type": "rate_limit_error",
    "status": 429,
    "details": {
      "retry_after": 60
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

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:
        error_data = response.json()
        error = error_data.get("error", {})
        details = error.get("details", {})
        
        # Get retry delay from details or header
        retry_after = details.get('retry_after')
        if not retry_after:
            retry_after = response.headers.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

{
  "success": false,
  "error": {
    "code": "INSUFFICIENT_PERMISSIONS",
    "message": "You don't have permission to perform this action.",
    "system_message": "User lacks required permissions",
    "type": "authorization_error",
    "status": 403,
    "details": {
      "required_role": "admin"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

{
  "success": false,
  "error": {
    "code": "THREAD_NOT_FOUND",
    "message": "The requested resource could not be found.",
    "system_message": "Thread not found",
    "type": "client_error",
    "status": 404,
    "details": {
      "thread_id": "thread_abc123"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

{
  "success": false,
  "error": {
    "code": "USER_ALREADY_EXISTS",
    "message": "An account with this email already exists.",
    "system_message": "Email already registered",
    "type": "client_error",
    "status": 409,
    "details": {
      "email": "user@example.com"
    },
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

{
  "success": false,
  "error": {
    "code": "INTERNAL_ERROR",
    "message": "Something went wrong. Please try again later.",
    "system_message": "Internal server error",
    "type": "server_error",
    "status": 500,
    "trace_id": "2fbbf3b6-51a1-4f1b-88e2-c00e8b52fbb8",
    "timestamp": "2025-01-15T10:30:00Z"
  }
}

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 = {
                "success": False,
                "error": {
                    "code": "UNKNOWN_ERROR",
                    "message": "Unknown error",
                    "status": response.status_code
                }
            }
        
        error = error_data.get("error", {})
        
        error_info = {
            "status_code": response.status_code,
            "error_code": error.get("code", "UNKNOWN_ERROR"),
            "error_type": error.get("type", "unknown_error"),
            "message": error.get("message", "Unknown error"),
            "system_message": error.get("system_message", "Unknown error"),
            "details": error.get("details", {}),
            "trace_id": error.get("trace_id"),
            "timestamp": error.get("timestamp"),
            "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)
        }
        
        # 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: Dict) -> str:
        """Get suggested action for error."""
        error_code = error.get("code", "")
        
        if status_code == 401:
            if "API_KEY" in error_code:
                return "Check API key validity and regenerate if necessary"
            elif "TOKEN" in error_code:
                return "Refresh authentication token or sign in again"
            return "Check authentication credentials"
        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['message']}"
        
        extra = {
            "status_code": error_info["status_code"],
            "error_code": error_info["error_code"],
            "error_type": error_info["error_type"],
            "trace_id": error_info["trace_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."""
        details = error_info.get("details", {})
        retry_after = details.get('retry_after') or 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: Dict) -> Dict[str, Any]:
        """Handle validation errors."""
        details = error_info.get("details", {})
        
        return {
            **error_info,
            "validation_details": details,
            "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['error_code']}): {error_info['message']}")
        
        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