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": "thrd_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.aitronos.com/v1/models",
 headers={"X-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 = {"X-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.aitronos.com/v1/{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.aitronos.com/v1/{endpoint}",
 headers={"X-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