Error Handling

The Functor SDK provides a comprehensive error handling system with detailed exception types, automatic retry logic, and graceful degradation patterns for robust application development.

Exception Hierarchy

Base Exception Classes

from functor_sdk import (
    # Base exceptions
    FunctorError,                    # Base exception for all SDK errors
    FunctorAPIError,                 # Base exception for API-related errors
    
    # Authentication & Authorization
    FunctorAuthenticationError,      # Invalid API key or token
    FunctorAuthorizationError,      # Insufficient permissions
    
    # Client errors
    FunctorValidationError,          # Invalid request parameters
    FunctorNotFoundError,           # Resource not found
    FunctorTimeoutError,            # Request timeout
    FunctorConnectionError,         # Network connection issues
    
    # Server errors
    FunctorRateLimitError,          # Rate limit exceeded
    FunctorServerError,             # Internal server error
    FunctorServiceUnavailableError  # Service temporarily unavailable
)

Exception Hierarchy Tree

FunctorError
├── FunctorAPIError
│   ├── FunctorAuthenticationError
│   ├── FunctorAuthorizationError
│   ├── FunctorValidationError
│   ├── FunctorNotFoundError
│   ├── FunctorTimeoutError
│   ├── FunctorConnectionError
│   ├── FunctorRateLimitError
│   ├── FunctorServerError
│   └── FunctorServiceUnavailableError
└── FunctorClientError  # Non-API client errors

Basic Error Handling

Simple Try-Catch Pattern

from functor_sdk import FunctorClient, FunctorAPIError

client = FunctorClient()

try:
    result = client.queries.execute("What is machine learning?")
    print(result.answer)
    
except FunctorAPIError as e:
    print(f"API Error {e.status_code}: {e.message}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Specific Exception Handling

from functor_sdk import (
    FunctorClient,
    FunctorAuthenticationError,
    FunctorNotFoundError,
    FunctorTimeoutError,
    FunctorConnectionError
)

client = FunctorClient()

try:
    result = client.queries.execute("What is AI?")
    print(result.answer)
    
except FunctorAuthenticationError:
    print("Authentication failed - check your API key")
    
except FunctorNotFoundError as e:
    print(f"Resource not found: {e.message}")
    
except FunctorTimeoutError:
    print("Request timed out - try again")
    
except FunctorConnectionError:
    print("Connection failed - check your network")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Advanced Error Handling Patterns

Retry Logic with Exponential Backoff

import time
from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorConnectionError,
    FunctorTimeoutError,
    FunctorRateLimitError
)

def execute_with_retry(client, operation, max_retries=3):
    for attempt in range(max_retries):
        try:
            return operation()
            
        except FunctorRateLimitError as e:
            if attempt < max_retries - 1:
                # Wait longer for rate limit
                wait_time = 2 ** (attempt + 2)  # 4, 8, 16 seconds
                print(f"Rate limited, waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
                
        except (FunctorConnectionError, FunctorTimeoutError) as e:
            if attempt < max_retries - 1:
                # Exponential backoff for connection issues
                wait_time = 2 ** attempt  # 1, 2, 4 seconds
                print(f"Connection error, retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
                
        except FunctorAPIError as e:
            # Don't retry for client errors (4xx)
            if 400 <= e.status_code < 500:
                raise
            # Retry for server errors (5xx)
            elif attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Server error, retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise

# Usage
client = FunctorClient()

def query_operation():
    return client.queries.execute("What is machine learning?")

try:
    result = execute_with_retry(client, query_operation)
    print(result.answer)
except Exception as e:
    print(f"Operation failed after retries: {e}")

Circuit Breaker Pattern

import time
from enum import Enum
from functor_sdk import FunctorClient, FunctorAPIError

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Circuit is open, failing fast
    HALF_OPEN = "half_open"  # Testing if service is back

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
    
    def call(self, operation):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = operation()
            self.on_success()
            return result
            
        except FunctorAPIError as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
client = FunctorClient()
circuit_breaker = CircuitBreaker(failure_threshold=3, timeout=30)

def query_operation():
    return client.queries.execute("What is AI?")

try:
    result = circuit_breaker.call(query_operation)
    print(result.answer)
except Exception as e:
    print(f"Circuit breaker error: {e}")

Graceful Degradation

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorConnectionError,
    FunctorTimeoutError
)

class GracefulClient:
    def __init__(self, client):
        self.client = client
        self.cache = {}  # Simple in-memory cache
    
    def execute_query_with_fallback(self, query):
        # Try primary operation
        try:
            result = self.client.queries.execute(query)
            # Cache successful result
            self.cache[query] = result
            return result
            
        except FunctorConnectionError:
            print("Connection failed, trying cached result...")
            return self._get_cached_result(query)
            
        except FunctorTimeoutError:
            print("Request timed out, trying cached result...")
            return self._get_cached_result(query)
            
        except FunctorAPIError as e:
            if e.status_code >= 500:  # Server error
                print("Server error, trying cached result...")
                return self._get_cached_result(query)
            else:  # Client error
                raise
    
    def _get_cached_result(self, query):
        if query in self.cache:
            cached_result = self.cache[query]
            print(f"Using cached result for: {query}")
            return cached_result
        else:
            raise Exception("No cached result available and service is unavailable")

# Usage
client = FunctorClient()
graceful_client = GracefulClient(client)

try:
    result = graceful_client.execute_query_with_fallback("What is machine learning?")
    print(result.answer)
except Exception as e:
    print(f"All fallback options failed: {e}")

Error Context and Debugging

Detailed Error Information

from functor_sdk import FunctorClient, FunctorAPIError
import traceback

client = FunctorClient()

try:
    result = client.queries.execute("What is AI?")
    
except FunctorAPIError as e:
    print("=== API Error Details ===")
    print(f"Error Type: {type(e).__name__}")
    print(f"Status Code: {e.status_code}")
    print(f"Message: {e.message}")
    print(f"Request URL: {e.request_url}")
    print(f"Request Method: {e.request_method}")
    
    if hasattr(e, 'response_body'):
        print(f"Response Body: {e.response_body}")
    
    if hasattr(e, 'request_body'):
        print(f"Request Body: {e.request_body}")
    
    print("\n=== Stack Trace ===")
    traceback.print_exc()
    
except Exception as e:
    print(f"Unexpected error: {e}")
    print("\n=== Stack Trace ===")
    traceback.print_exc()

Error Logging

import logging
from functor_sdk import FunctorClient, FunctorAPIError

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class LoggingClient:
    def __init__(self, client):
        self.client = client
    
    def execute_query_with_logging(self, query):
        logger.info(f"Executing query: {query}")
        
        try:
            result = self.client.queries.execute(query)
            logger.info(f"Query successful: {len(result.answer)} characters")
            return result
            
        except FunctorAPIError as e:
            logger.error(f"API Error {e.status_code}: {e.message}")
            logger.error(f"Request URL: {e.request_url}")
            raise
            
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            raise

# Usage
client = FunctorClient()
logging_client = LoggingClient(client)

try:
    result = logging_client.execute_query_with_logging("What is machine learning?")
    print(result.answer)
except Exception as e:
    print(f"Query failed: {e}")

Error Recovery Strategies

Automatic Recovery

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorConnectionError,
    FunctorTimeoutError,
    FunctorRateLimitError
)

class ResilientClient:
    def __init__(self, client):
        self.client = client
        self.retry_config = {
            FunctorConnectionError: {"max_retries": 3, "backoff": "exponential"},
            FunctorTimeoutError: {"max_retries": 2, "backoff": "linear"},
            FunctorRateLimitError: {"max_retries": 5, "backoff": "exponential"},
        }
    
    def execute_with_recovery(self, operation, *args, **kwargs):
        last_exception = None
        
        for attempt in range(5):  # Maximum attempts
            try:
                return operation(*args, **kwargs)
                
            except Exception as e:
                last_exception = e
                
                # Check if we should retry this error
                retry_config = self.retry_config.get(type(e))
                if not retry_config or attempt >= retry_config["max_retries"]:
                    break
                
                # Calculate wait time
                if retry_config["backoff"] == "exponential":
                    wait_time = 2 ** attempt
                else:  # linear
                    wait_time = attempt + 1
                
                print(f"Attempt {attempt + 1} failed: {e}")
                print(f"Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
        
        # All retries failed
        raise last_exception

# Usage
client = FunctorClient()
resilient_client = ResilientClient(client)

try:
    result = resilient_client.execute_with_recovery(
        client.queries.execute,
        "What is machine learning?"
    )
    print(result.answer)
except Exception as e:
    print(f"Operation failed after all retries: {e}")

Health Check Integration

from functor_sdk import FunctorClient, FunctorAPIError

class HealthAwareClient:
    def __init__(self, client):
        self.client = client
        self.last_health_check = 0
        self.health_check_interval = 300  # 5 minutes
    
    def execute_with_health_check(self, operation, *args, **kwargs):
        # Check if we need to verify system health
        current_time = time.time()
        if current_time - self.last_health_check > self.health_check_interval:
            if not self._check_system_health():
                raise Exception("System health check failed")
            self.last_health_check = current_time
        
        # Execute operation
        try:
            return operation(*args, **kwargs)
        except FunctorAPIError as e:
            # If we get a server error, check system health
            if e.status_code >= 500:
                if not self._check_system_health():
                    raise Exception("System appears to be down")
            raise
    
    def _check_system_health(self):
        try:
            health = self.client.health.check()
            return health["status"] == "healthy"
        except Exception:
            return False

# Usage
client = FunctorClient()
health_client = HealthAwareClient(client)

try:
    result = health_client.execute_with_health_check(
        client.queries.execute,
        "What is AI?"
    )
    print(result.answer)
except Exception as e:
    print(f"Operation failed: {e}")

Best Practices

Error Handling Best Practices

Handle specific exceptions: Catch specific exception types rather than generic ones
Implement retry logic: For transient errors like network issues
Use exponential backoff: Avoid overwhelming the server
Log errors appropriately: Include context for debugging
Implement circuit breakers: Prevent cascading failures
Provide fallback mechanisms: Graceful degradation when possible
Monitor error rates: Track and alert on error patterns

Production Error Handling

from functor_sdk import FunctorClient, FunctorAPIError
import logging
import time
from typing import Optional, Callable, Any

class ProductionErrorHandler:
    def __init__(self, client: FunctorClient):
        self.client = client
        self.logger = logging.getLogger(__name__)
        self.error_counts = {}
        self.max_errors_per_minute = 10
    
    def execute_with_production_handling(
        self, 
        operation: Callable, 
        operation_name: str,
        *args, 
        **kwargs
    ) -> Optional[Any]:
        """Execute operation with production-grade error handling"""
        
        # Check error rate
        if self._is_error_rate_too_high():
            self.logger.warning(f"Error rate too high, skipping {operation_name}")
            return None
        
        try:
            result = operation(*args, **kwargs)
            self._reset_error_count(operation_name)
            return result
            
        except FunctorAPIError as e:
            self._handle_api_error(e, operation_name)
            return None
            
        except Exception as e:
            self._handle_unexpected_error(e, operation_name)
            return None
    
    def _is_error_rate_too_high(self) -> bool:
        current_minute = int(time.time() / 60)
        total_errors = sum(
            count for minute, count in self.error_counts.items()
            if minute >= current_minute - 1
        )
        return total_errors > self.max_errors_per_minute
    
    def _handle_api_error(self, error: FunctorAPIError, operation_name: str):
        self._increment_error_count(operation_name)
        
        if error.status_code >= 500:
            self.logger.error(f"Server error in {operation_name}: {error.message}")
        elif error.status_code == 429:
            self.logger.warning(f"Rate limited in {operation_name}")
        elif error.status_code >= 400:
            self.logger.warning(f"Client error in {operation_name}: {error.message}")
    
    def _handle_unexpected_error(self, error: Exception, operation_name: str):
        self._increment_error_count(operation_name)
        self.logger.error(f"Unexpected error in {operation_name}: {error}")
    
    def _increment_error_count(self, operation_name: str):
        current_minute = int(time.time() / 60)
        key = f"{operation_name}_{current_minute}"
        self.error_counts[key] = self.error_counts.get(key, 0) + 1
    
    def _reset_error_count(self, operation_name: str):
        # Clean up old error counts
        current_minute = int(time.time() / 60)
        self.error_counts = {
            k: v for k, v in self.error_counts.items()
            if int(k.split('_')[-1]) >= current_minute - 1
        }

# Usage
client = FunctorClient()
error_handler = ProductionErrorHandler(client)

result = error_handler.execute_with_production_handling(
    client.queries.execute,
    "query_execution",
    "What is machine learning?"
)

if result:
    print(result.answer)
else:
    print("Query failed, check logs for details")

Next Steps

Async vs Sync - Choose the right interface for error handling
Client Reference - Configure client for robust error handling
Health Namespace - Monitor system health
Queries Namespace - Apply error handling to queries