Skip to content

Latest commit

 

History

History
622 lines (481 loc) · 16.7 KB

File metadata and controls

622 lines (481 loc) · 16.7 KB

Error Handling

The Stratix Python SDK provides a comprehensive exception hierarchy to help you handle different error conditions gracefully. This guide covers all available exception types and best practices for error handling.

Exception Hierarchy

All Stratix exceptions inherit from the base StratixError class:

StratixError
├── APIError
│   ├── APIConnectionError
│   │   └── APITimeoutError
│   ├── APIResponseValidationError
│   └── APIStatusError
│       ├── BadRequestError (400)
│       ├── AuthenticationError (401)
│       ├── PermissionDeniedError (403)
│       ├── NotFoundError (404)
│       ├── ConflictError (409)
│       ├── UnprocessableEntityError (422)
│       ├── RateLimitError (429)
│       └── InternalServerError (500+)

Exception Types

Base Exceptions

StratixError

Base exception for all Stratix-related errors.

import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.StratixError as e:
    print(f"Stratix error occurred: {e}")

APIError

Base exception for all API-related errors. Contains additional context about the request.

Properties:

  • message: Error message
  • request: The HTTP request that caused the error
  • body: Response body (if available)
import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APIError as e:
    print(f"API error: {e.message}")
    print(f"Request URL: {e.request.url}")
    print(f"Response body: {e.body}")

Connection Errors

APIConnectionError

Raised when the client cannot connect to the API server.

Common causes:

  • Network connectivity issues
  • DNS resolution problems
  • Server is down
  • Firewall blocking requests
import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APIConnectionError as e:
    print("Connection failed - check your network connection")
    print(f"Error details: {e}")

APITimeoutError

Raised when a request times out.

import layerlens

try:
    client = layerlens.Stratix(timeout=0.2)  # Very short timeout
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APITimeoutError:
    print("Request timed out - try increasing timeout or check network")

HTTP Status Errors

All HTTP status errors inherit from APIStatusError and include additional properties:

Properties:

  • status_code: HTTP status code
  • response: Full HTTP response object
  • request_id: Request ID for tracking (if provided by server)

BadRequestError (400)

Request was malformed or contained invalid parameters.

import layerlens

try:
    client = layerlens.Stratix()
    # Invalid parameters
    evaluation = client.evaluations.create(model="", benchmark="")
except layerlens.BadRequestError as e:
    print(f"Bad request: {e}")
    print(f"Status code: {e.status_code}")

AuthenticationError (401)

API key is missing, invalid, or expired.

import layerlens

try:
    client = layerlens.Stratix(api_key="invalid_key")
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.AuthenticationError:
    print("Authentication failed - check your API key")
    print("Make sure LAYERLENS_STRATIX_API_KEY is set correctly")

PermissionDeniedError (403)

Valid API key but insufficient permissions for the requested operation.

import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="restricted-model", benchmark="mmlu")
except layerlens.PermissionDeniedError:
    print("Permission denied - check your organization/project access")
    print("Contact your administrator for access to this resource")

NotFoundError (404)

Requested resource (model, benchmark, evaluation) does not exist.

import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="nonexistent-model", benchmark="mmlu")
except layerlens.NotFoundError:
    print("Model or benchmark not found")
    print("Check available models and benchmarks in the Stratix Dashboard")

ConflictError (409)

Request conflicts with current resource state.

import layerlens

try:
    client = layerlens.Stratix()
    # Some operation that conflicts with current state
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.ConflictError:
    print("Request conflicts with current state")

UnprocessableEntityError (422)

Request parameters are valid but cannot be processed.

import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="invalid-benchmark")
except layerlens.UnprocessableEntityError as e:
    print(f"Cannot process request: {e}")
    print("Parameters are valid but operation cannot be completed")

RateLimitError (429)

Too many requests sent in a given time period.

import layerlens
import time

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.RateLimitError as e:
    print("Rate limit exceeded")
    # Extract retry-after header if available
    retry_after = e.response.headers.get('retry-after')
    if retry_after:
        print(f"Retry after {retry_after} seconds")
        time.sleep(int(retry_after))
    else:
        print("Waiting 60 seconds before retry...")
        time.sleep(60)

InternalServerError (500+)

Server-side error occurred.

import layerlens

try:
    client = layerlens.Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.InternalServerError as e:
    print(f"Server error: {e.status_code}")
    print("This is a server-side issue - try again later")
    print(f"Request ID: {e.request_id}")  # For support tickets

Best Practices

1. Handle Specific Exceptions

import layerlens
import time
from layerlens import Stratix

def robust_create_evaluation(model: str, benchmark: str, max_retries: int = 3):
    client = Stratix()

    for attempt in range(max_retries):
        try:
            evaluation = client.evaluations.create(model=model, benchmark=benchmark)
            return evaluation

        except layerlens.AuthenticationError:
            print("❌ Authentication failed - check your API key")
            break  # Don't retry auth errors

        except layerlens.PermissionDeniedError:
            print("❌ Permission denied - contact your administrator")
            break  # Don't retry permission errors

        except layerlens.NotFoundError:
            print(f"❌ Model '{model}' or benchmark '{benchmark}' not found")
            break  # Don't retry not found errors

        except layerlens.RateLimitError as e:
            retry_after = e.response.headers.get('retry-after', 60)
            print(f"⏳ Rate limited - waiting {retry_after} seconds...")
            time.sleep(int(retry_after))
            continue  # Retry after waiting

        except layerlens.InternalServerError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"🔄 Server error - retrying in {wait_time}s (attempt {attempt + 1})")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Server error - max retries exceeded")
                break

        except layerlens.APIConnectionError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"🔄 Connection error - retrying in {wait_time}s (attempt {attempt + 1})")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Connection failed - check your network")
                break

        except layerlens.APIError as e:
            print(f"❌ Unexpected API error: {e}")
            break

    return None

2. Graceful Degradation

import layerlens
from layerlens import Stratix

def get_evaluation_results_with_fallback(evaluation_id: str):
    client = Stratix()

    try:
        results = client.results.get(evaluation_id=evaluation_id)

        if results:
            return {"success": True, "data": results, "message": "Results retrieved successfully"}
        else:
            return {"success": False, "data": None, "message": "No results found"}

    except layerlens.NotFoundError:
        return {"success": False, "data": None, "message": "Evaluation not found"}

    except layerlens.AuthenticationError:
        return {"success": False, "data": None, "message": "Authentication required"}

    except layerlens.APIConnectionError:
        return {"success": False, "data": None, "message": "Service temporarily unavailable"}

    except layerlens.APIError as e:
        return {"success": False, "data": None, "message": f"Service error: {e}"}

# Usage
result = get_evaluation_results_with_fallback("eval_123")
if result["success"]:
    process_results(result["data"])
else:
    print(f"Could not get results: {result['message']}")

3. Logging and Monitoring

import logging
import layerlens
from layerlens import Stratix

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_api_call():
    client = Stratix()

    try:
        logger.info("Creating evaluation...")
        evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")

        if evaluation:
            logger.info(f"Evaluation created successfully: {evaluation.id}")
            return evaluation
        else:
            logger.warning("Evaluation creation returned None")
            return None

    except layerlens.RateLimitError as e:
        logger.warning(f"Rate limited - request ID: {e.request_id}")
        raise

    except layerlens.AuthenticationError:
        logger.error("Authentication failed - check API key configuration")
        raise

    except layerlens.APIConnectionError:
        logger.error("Network connection failed")
        raise

    except layerlens.InternalServerError as e:
        logger.error(f"Server error: {e.status_code} - request ID: {e.request_id}")
        raise

    except layerlens.APIError as e:
        logger.error(f"Unexpected API error: {e} - request ID: {getattr(e, 'request_id', 'N/A')}")
        raise

4. Context Managers for Resource Management

import layerlens
from contextlib import contextmanager
from layerlens import Stratix

@contextmanager
def stratix_client():
    """Context manager for Stratix client with error handling"""
    client = None
    try:
        client = Stratix()
        yield client
    except layerlens.AuthenticationError:
        print("Authentication failed")
        raise
    except layerlens.APIConnectionError:
        print("Connection failed")
        raise
    finally:
        # Cleanup if needed
        pass

# Usage
try:
    with stratix_client() as client:
        evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
        results = client.results.get(evaluation_id=evaluation.id)
except layerlens.StratixError:
    print("Stratix operation failed")

Error Response Details

Status Error Properties

import layerlens
from layerlens import Stratix

try:
    client = Stratix()
    evaluation = client.evaluations.create(model="invalid", benchmark="invalid")
except layerlens.APIStatusError as e:
    print(f"Status Code: {e.status_code}")
    print(f"Request ID: {e.request_id}")
    print(f"Response Headers: {dict(e.response.headers)}")
    print(f"Response Body: {e.body}")
    print(f"Request URL: {e.request.url}")
    print(f"Request Method: {e.request.method}")

Extracting Useful Information

import layerlens
from layerlens import Stratix

def extract_error_info(error: layerlens.APIError):
    info = {
        "type": type(error).__name__,
        "message": str(error),
        "request_url": error.request.url if hasattr(error, 'request') else None,
        "request_method": error.request.method if hasattr(error, 'request') else None,
    }

    if hasattr(error, 'status_code'):
        info["status_code"] = error.status_code

    if hasattr(error, 'request_id'):
        info["request_id"] = error.request_id

    if hasattr(error, 'response'):
        info["response_headers"] = dict(error.response.headers)

    return info

# Usage
try:
    client = Stratix()
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APIError as e:
    error_info = extract_error_info(e)
    print(f"Error details: {error_info}")

Testing Error Handling

import pytest
import layerlens
from unittest.mock import Mock, patch
from layerlens import Stratix

def test_authentication_error_handling():
    """Test that authentication errors are handled properly"""
    with patch('layerlens.Stratix') as mock_stratix:
        mock_stratix.side_effect = layerlens.AuthenticationError(
            "Invalid API key",
            request=Mock(),
            response=Mock()
        )

        with pytest.raises(layerlens.AuthenticationError):
            client = Stratix()
            client.evaluations.create(model="gpt-4", benchmark="mmlu")

def test_rate_limit_retry():
    """Test that rate limit errors trigger appropriate retry logic"""
    # Your retry logic test here
    pass

Common Error Scenarios

Invalid Configuration

# Missing API key
try:
    client = Stratix(api_key=None)
except layerlens.StratixError as e:
    print(f"Configuration error: {e}")

Network Issues

# Connection timeout
try:
    client = Stratix(timeout=0.1)  # Very short timeout
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APITimeoutError:
    print("Request timed out")

# Network connectivity
try:
    # Simulate network issues
    evaluation = client.evaluations.create(model="gpt-4", benchmark="mmlu")
except layerlens.APIConnectionError:
    print("Network connectivity issue")

Error Recovery Strategies

Exponential Backoff

import time
import random
import layerlens
from layerlens import Stratix

def exponential_backoff_retry(func, max_retries=3, base_delay=1):
    """Retry function with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return func()
        except (layerlens.InternalServerError, layerlens.APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise

            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed, retrying in {delay:.2f}s...")
            time.sleep(delay)

# Usage
def create_evaluation():
    client = Stratix()
    return client.evaluations.create(model="gpt-4", benchmark="mmlu")

evaluation = exponential_backoff_retry(create_evaluation)

Circuit Breaker Pattern

import time
from enum import Enum
from layerlens import Stratix
import layerlens

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

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, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time < self.timeout:
                raise layerlens.APIConnectionError(message="Circuit breaker is OPEN")
            else:
                self.state = CircuitState.HALF_OPEN

        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except (layerlens.InternalServerError, layerlens.APIConnectionError) 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
breaker = CircuitBreaker()
client = Stratix()

try:
    evaluation = breaker.call(
        client.evaluations.create,
        model="gpt-4",
        benchmark="mmlu"
    )
except layerlens.APIError as e:
    print(f"Circuit breaker prevented call or operation failed: {e}")