Chapter 10: Error Handling

{
  "error": "Error Type",
  "message": "Detailed human-readable error message",
  "request_id": "req-123e4567-e89b-12d3-a456-426614174000",
  "timestamp": "2024-01-15T12:00:00Z"
}

{
  "error": "Validation Error",
  "message": "Request validation failed",
  "details": [
    {
      "field": "email",
      "message": "Invalid email format"
    },
    {
      "field": "password",
      "message": "Password must be at least 8 characters"
    }
  ],
  "request_id": "req-123e4567-e89b-12d3-a456-426614174000",
  "timestamp": "2024-01-15T12:00:00Z"
}

{
  "error": "Missing API key",
  "message": "Expected 'X-API-Key' or 'Authorization: Bearer' header with access key"
}

# Ensure you include the API key
headers = {
    'X-API-Key': 'ps_live_your_key_here',
    'Content-Type': 'application/json'
}

{
  "error": "Invalid API key",
  "message": "The provided access key is not valid"
}

# Verify your key
curl -X GET https://api.polysystems.ai/api/keys \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  | jq '.[] | select(.key_value_preview | startswith("ps_live_abc"))'

{
  "error": "API key expired",
  "message": "This access key has expired"
}

# Generate new token
def refresh_api_key():
    response = requests.post(
        'https://api.polysystems.ai/api/keys',
        headers={'Authorization': f'Bearer {jwt_token}'},
        json={'name': 'Replacement Token'}
    )
    return response.json()['key_value']

{
  "error": "Insufficient credits",
  "message": "This request costs $0.0020 but your balance is insufficient. Please top up your account."
}

def check_balance_and_topup(required_amount):
    response = requests.get(
        'https://api.polysystems.ai/api/payments/credits/balance',
        headers={'Authorization': f'Bearer {jwt_token}'}
    )
    balance = response.json()['balance']
    
    if balance < required_amount:
        print(f"Balance too low: ${balance}. Please add credits.")
        return False
    return True

{
  "error": "Spending limit exceeded",
  "message": "This access key has reached its spending limit. Please adjust limits or top up credits."
}

# Check and reset limits
def handle_limit_exceeded(key_id):
    # Check current limits
    response = requests.get(
        f'https://api.polysystems.ai/api/keys/{key_id}/limits',
        headers={'Authorization': f'Bearer {jwt_token}'}
    )
    limits = response.json()
    
    print(f"Daily: ${limits['daily_spent']:.2f} / ${limits['daily_limit']:.2f}")
    print(f"Monthly: ${limits['monthly_spent']:.2f} / ${limits['monthly_limit']:.2f}")
    
    # Option 1: Wait for reset
    # Option 2: Increase limits
    # Option 3: Use different token

{
  "error": "Rate limit exceeded",
  "message": "You have exceeded the rate limit. Please retry after 60 seconds.",
  "retry_after": 60,
  "limit": 1000,
  "window": "hour"
}

import time
 
def make_request_with_rate_limit_handling(url, headers, data):
    response = requests.post(url, headers=headers, json=data)
    
    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        print(f"Rate limited. Waiting {retry_after}s...")
        time.sleep(retry_after)
        return make_request_with_rate_limit_handling(url, headers, data)
    
    return response

{
  "error": "Validation Error",
  "message": "Request validation failed",
  "details": [
    {
      "field": "messages",
      "message": "messages is required"
    }
  ]
}

def validate_request_data(data):
    """Validate request before sending"""
    errors = []
    
    if 'messages' not in data:
        errors.append({'field': 'messages', 'message': 'messages is required'})
    
    if not isinstance(data.get('messages', []), list):
        errors.append({'field': 'messages', 'message': 'messages must be an array'})
    
    if errors:
        raise ValueError(f"Validation failed: {errors}")
    
    return True

{
  "error": "Not Found",
  "message": "The requested resource was not found"
}

def safe_get_resource(resource_id):
    response = requests.get(
        f'https://api.polysystems.ai/api/keys/{resource_id}',
        headers={'Authorization': f'Bearer {jwt_token}'}
    )
    
    if response.status_code == 404:
        print(f"Resource {resource_id} not found")
        return None
    
    response.raise_for_status()
    return response.json()

{
  "error": "Internal Server Error",
  "message": "An unexpected error occurred. Please try again later.",
  "request_id": "req-123e4567-e89b-12d3-a456-426614174000"
}

def make_request_with_retry(url, headers, data, max_retries=3):
    """Retry on server errors"""
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data, timeout=30)
            
            # Retry on server errors
            if response.status_code >= 500:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # Exponential backoff
                    print(f"Server error. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
                    continue
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                print(f"Request timeout. Retrying...")
                time.sleep(2 ** attempt)
                continue
            raise
    
    raise Exception("Max retries exceeded")

import requests
import time
import logging
from typing import Optional, Dict, Any
 
logger = logging.getLogger(__name__)
 
class APIError(Exception):
    """Base exception for API errors"""
    def __init__(self, status_code, error_type, message, request_id=None):
        self.status_code = status_code
        self.error_type = error_type
        self.message = message
        self.request_id = request_id
        super().__init__(f"{error_type}: {message}")
 
class APIClient:
    def __init__(self, api_key: str, base_url: str = "https://api.polysystems.ai"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': api_key,
            'Content-Type': 'application/json'
        })
    
    def _handle_response(self, response: requests.Response) -> Dict[Any, Any]:
        """Handle API response and errors"""
        # Success
        if 200 <= response.status_code < 300:
            return response.json()
        
        # Parse error response
        try:
            error_data = response.json()
            error_type = error_data.get('error', 'Unknown Error')
            message = error_data.get('message', 'No error message provided')
            request_id = error_data.get('request_id')
        except ValueError:
            error_type = f"HTTP {response.status_code}"
            message = response.text
            request_id = None
        
        # Log error
        logger.error(
            f"API Error: {error_type} - {message}",
            extra={
                'status_code': response.status_code,
                'request_id': request_id,
                'url': response.url
            }
        )
        
        # Raise appropriate exception
        raise APIError(response.status_code, error_type, message, request_id)
    
    def request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict] = None,
        params: Optional[Dict] = None,
        max_retries: int = 3
    ) -> Dict[Any, Any]:
        """Make API request with error handling and retries"""
        url = f"{self.base_url}{endpoint}"
        
        for attempt in range(max_retries):
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    json=data,
                    params=params,
                    timeout=30
                )
                
                # Handle rate limiting
                if response.status_code == 429:
                    retry_after = int(response.headers.get('Retry-After', 60))
                    if attempt < max_retries - 1:
                        logger.warning(f"Rate limited. Waiting {retry_after}s...")
                        time.sleep(retry_after)
                        continue
                
                # Handle server errors with retry
                if response.status_code >= 500:
                    if attempt < max_retries - 1:
                        wait_time = 2 ** attempt
                        logger.warning(f"Server error. Retrying in {wait_time}s...")
                        time.sleep(wait_time)
                        continue
                
                return self._handle_response(response)
                
            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    logger.warning(f"Request timeout. Retrying...")
                    time.sleep(2 ** attempt)
                    continue
                raise
            
            except requests.exceptions.ConnectionError as e:
                if attempt < max_retries - 1:
                    logger.warning(f"Connection error. Retrying...")
                    time.sleep(2 ** attempt)
                    continue
                raise
        
        raise Exception("Max retries exceeded")
    
    def post(self, endpoint: str, data: Dict) -> Dict:
        """POST request"""
        return self.request('POST', endpoint, data=data)
    
    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
        """GET request"""
        return self.request('GET', endpoint, params=params)
    
    def put(self, endpoint: str, data: Dict) -> Dict:
        """PUT request"""
        return self.request('PUT', endpoint, data=data)
    
    def delete(self, endpoint: str) -> Dict:
        """DELETE request"""
        return self.request('DELETE', endpoint)
 
# Usage
try:
    client = APIClient(api_key='ps_live_your_key_here')
    
    result = client.post('/api/hub/agents/chat', {
        'messages': [
            {'role': 'user', 'content': 'Hello!'}
        ]
    })
    
    print(result)
    
except APIError as e:
    if e.status_code == 401:
        print("Authentication failed. Check your API key.")
    elif e.status_code == 402:
        print("Insufficient credits. Please add funds.")
    elif e.status_code == 429:
        print("Rate limit exceeded. Slow down requests.")
    else:
        print(f"API error: {e.message}")

const axios = require('axios');
 
class APIError extends Error {
  constructor(statusCode, errorType, message, requestId) {
    super(`${errorType}: ${message}`);
    this.statusCode = statusCode;
    this.errorType = errorType;
    this.requestId = requestId;
    this.name = 'APIError';
  }
}
 
class APIClient {
  constructor(apiKey, baseURL = 'https://api.polysystems.ai') {
    this.apiKey = apiKey;
    this.baseURL = baseURL;
    this.client = axios.create({
      baseURL: baseURL,
      headers: {
        'X-API-Key': apiKey,
        'Content-Type': 'application/json'
      },
      timeout: 30000
    });
  }
 
  async request(method, endpoint, data = null, maxRetries = 3) {
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      try {
        const response = await this.client.request({
          method,
          url: endpoint,
          data
        });
        
        return response.data;
        
      } catch (error) {
        if (error.response) {
          // Server responded with error status
          const { status, data } = error.response;
          const errorType = data.error || 'Unknown Error';
          const message = data.message || 'No error message';
          const requestId = data.request_id;
          
          // Handle rate limiting
          if (status === 429 && attempt < maxRetries - 1) {
            const retryAfter = parseInt(error.response.headers['retry-after'] || 60);
            console.warn(`Rate limited. Waiting ${retryAfter}s...`);
            await this.sleep(retryAfter * 1000);
            continue;
          }
          
          // Handle server errors
          if (status >= 500 && attempt < maxRetries - 1) {
            const waitTime = Math.pow(2, attempt);
            console.warn(`Server error. Retrying in ${waitTime}s...`);
            await this.sleep(waitTime * 1000);
            continue;
          }
          
          throw new APIError(status, errorType, message, requestId);
          
        } else if (error.code === 'ECONNABORTED' || error.code === 'ETIMEDOUT') {
          // Timeout
          if (attempt < maxRetries - 1) {
            console.warn('Request timeout. Retrying...');
            await this.sleep(Math.pow(2, attempt) * 1000);
            continue;
          }
          throw error;
          
        } else {
          // Network error
          if (attempt < maxRetries - 1) {
            console.warn('Network error. Retrying...');
            await this.sleep(Math.pow(2, attempt) * 1000);
            continue;
          }
          throw error;
        }
      }
    }
    
    throw new Error('Max retries exceeded');
  }
 
  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
 
  async post(endpoint, data) {
    return this.request('POST', endpoint, data);
  }
 
  async get(endpoint) {
    return this.request('GET', endpoint);
  }
 
  async put(endpoint, data) {
    return this.request('PUT', endpoint, data);
  }
 
  async delete(endpoint) {
    return this.request('DELETE', endpoint);
  }
}
 
// Usage
(async () => {
  try {
    const client = new APIClient('ps_live_your_key_here');
    
    const result = await client.post('/api/hub/agents/chat', {
      messages: [
        { role: 'user', content: 'Hello!' }
      ]
    });
    
    console.log(result);
    
  } catch (error) {
    if (error instanceof APIError) {
      if (error.statusCode === 401) {
        console.error('Authentication failed. Check your API key.');
      } else if (error.statusCode === 402) {
        console.error('Insufficient credits. Please add funds.');
      } else if (error.statusCode === 429) {
        console.error('Rate limit exceeded. Slow down requests.');
      } else {
        console.error(`API error: ${error.message}`);
      }
    } else {
      console.error(`Unexpected error: ${error.message}`);
    }
  }
})();

import logging
import json
from datetime import datetime
 
class APIErrorLogger:
    def __init__(self, log_file='api_errors.log'):
        self.logger = logging.getLogger('api_errors')
        self.logger.setLevel(logging.ERROR)
        
        handler = logging.FileHandler(log_file)
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
    
    def log_error(self, error_type, message, context=None):
        """Log error in structured format"""
        log_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'error_type': error_type,
            'message': message,
            'context': context or {}
        }
        
        self.logger.error(json.dumps(log_entry))
 
# Usage
error_logger = APIErrorLogger()
 
try:
    response = make_api_call()
except APIError as e:
    error_logger.log_error(
        error_type=e.error_type,
        message=e.message,
        context={
            'status_code': e.status_code,
            'request_id': e.request_id,
            'user_id': current_user_id
        }
    )

from collections import defaultdict
from datetime import datetime, timedelta
 
class ErrorMetrics:
    def __init__(self):
        self.errors = defaultdict(int)
        self.error_times = []
    
    def record_error(self, error_type, status_code):
        """Record error occurrence"""
        self.errors[f"{status_code}_{error_type}"] += 1
        self.error_times.append(datetime.utcnow())
    
    def get_error_rate(self, window_minutes=60):
        """Get error rate in time window"""
        cutoff = datetime.utcnow() - timedelta(minutes=window_minutes)
        recent_errors = [t for t in self.error_times if t > cutoff]
        return len(recent_errors) / window_minutes  # Errors per minute
    
    def get_most_common_errors(self, top=5):
        """Get most common error types"""
        sorted_errors = sorted(
            self.errors.items(),
            key=lambda x: x[1],
            reverse=True
        )
        return sorted_errors[:top]
 
# Usage
metrics = ErrorMetrics()
 
try:
    result = make_api_call()
except APIError as e:
    metrics.record_error(e.error_type, e.status_code)
    
    # Check if error rate is too high
    error_rate = metrics.get_error_rate(window_minutes=10)
    if error_rate > 10:  # More than 10 errors per minute
        send_alert(f"High error rate: {error_rate:.2f} errors/min")

# ✅ Good: Proper error handling
try:
    result = client.post('/api/hub/agents/chat', data)
    process_result(result)
except APIError as e:
    if e.status_code == 402:
        notify_insufficient_credits()
    elif e.status_code == 429:
        schedule_retry_later()
    else:
        log_error_and_notify(e)
 
# ❌ Bad: Ignoring errors
result = client.post('/api/hub/agents/chat', data)
process_result(result)  # Crashes on error

from datetime import datetime, timedelta
 
class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout_seconds=60):
        self.failure_threshold = failure_threshold
        self.timeout = timedelta(seconds=timeout_seconds)
        self.failures = 0
        self.last_failure = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN
    
    def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker"""
        if self.state == 'OPEN':
            if datetime.utcnow() - self.last_failure > self.timeout:
                self.state = 'HALF_OPEN'
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        """Reset on success"""
        self.failures = 0
        self.state = 'CLOSED'
    
    def on_failure(self):
        """Increment failures"""
        self.failures += 1
        self.last_failure = datetime.utcnow()
        
        if self.failures >= self.failure_threshold:
            self.state = 'OPEN'
 
# Usage
breaker = CircuitBreaker(failure_threshold=5, timeout_seconds=60)
 
try:
    result = breaker.call(make_api_call, data)
except Exception as e:
    print(f"Circuit breaker: {e}")

def get_ai_response_with_fallback(prompt):
    """Try API with fallback to cached/default response"""
    try:
        # Try primary API
        response = client.post('/api/hub/agents/chat', {
            'messages': [{'role': 'user', 'content': prompt}]
        })
        return response['message']
        
    except APIError as e:
        if e.status_code == 402:
            # Insufficient credits - use cached response
            return get_cached_response(prompt)
        elif e.status_code == 429:
            # Rate limited - use queue
            return queue_for_later(prompt)
        elif e.status_code >= 500:
            # Server error - use default response
            return "I'm temporarily unavailable. Please try again later."
        else:
            raise

ERROR_MESSAGES = {
    401: "Authentication failed. Please check your credentials and try again.",
    402: "You don't have enough credits. Please add funds to continue.",
    403: "You don't have permission to perform this action.",
    404: "The requested resource was not found.",
    429: "You're making requests too quickly. Please slow down.",
    500: "We're experiencing technical difficulties. Please try again later.",
    503: "The service is temporarily unavailable. Please try again in a few minutes."
}
 
def get_user_friendly_error(status_code, default=None):
    """Get user-friendly error message"""
    return ERROR_MESSAGES.get(
        status_code,
        default or "An unexpected error occurred. Please try again."
    )