Skip to main content

Error Handling

The SDK provides comprehensive error handling with rich exception classes that provide context about what went wrong and how to handle it.

Exception Hierarchy​

The SDK provides a comprehensive hierarchy of exception classes that map to different types of errors. Import the exceptions you need:

Import Exceptions
from nexla_sdk.exceptions import (
    NexlaAPIError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ResourceConflictError,
    ServerError
)

HTTP Error Mapping​

The SDK automatically maps HTTP status codes to appropriate exception classes, providing rich context about what went wrong:

  • 400 → ValidationError
  • 401 → AuthenticationError
  • 403 → AuthorizationError
  • 404 → NotFoundError
  • 409 → ResourceConflictError
  • 429 → RateLimitError (with retry_after if available)
  • 5xx → ServerError

Basic Error Handling​

Basic Error Handling
from nexla_sdk import NexlaClient
from nexla_sdk.exceptions import NexlaAPIError, AuthenticationError

client = NexlaClient()

try:
    # This might fail if flow doesn't exist
    flow = client.flows.get("non-existent-flow-id")
except AuthenticationError:
    print("❌ Authentication failed - check your credentials")
except NexlaAPIError as e:
    print(f"❌ API Error: {e.message}")
    print(f"Status Code: {e.status_code}")
except Exception as e:
    print(f"❌ Unexpected error: {e}")

Advanced Error Handling​

Safe API Calls with Retries​

Safe API Call with Retries
from nexla_sdk.exceptions import NexlaAPIError, AuthenticationError, RateLimitError
import time

def safe_api_call_with_retries(func, max_retries=3):
    """Execute API call with automatic retries"""
    for attempt in range(max_retries):
        try:
            return func()
        except AuthenticationError:
            print("❌ Authentication failed:")
            print("   • Check your NEXLA_SERVICE_KEY")
            print("   • Verify the key is active")
            print("   • Ensure you have proper permissions")
            raise  # Don't retry auth errors
        except RateLimitError as e:
            wait_time = e.retry_after or (2 ** attempt)  # Exponential backoff
            print(f"⏳ Rate limited. Waiting {wait_time}s before retry {attempt + 1}/{max_retries}")
            if attempt < max_retries - 1:
                time.sleep(wait_time)
                continue
            else:
                raise
        except NexlaAPIError as e:
            print(f"❌ API Error (Status {e.status_code}): {e.message}")
            if e.status_code >= 500 and attempt < max_retries - 1:
                # Retry server errors
                wait_time = 2 ** attempt
                print(f"⏳ Server error. Retrying in {wait_time}s...")
                time.sleep(wait_time)
                continue
            else:
                raise
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
            raise

# Usage example
def get_flows():
    return client.flows.list()

flows = safe_api_call_with_retries(get_flows)

Resource Conflict Handling​

Resource Conflict Handling
from nexla_sdk.exceptions import ResourceConflictError

def create_or_update_destination(payload):
    """Create destination or update if it already exists"""
    try:
        destination = client.destinations.create(payload)
        print(f"✅ Created new destination: {destination.name}")
        return destination
    except ResourceConflictError:
        print("⚠️  Destination already exists, updating instead...")
        # Find existing destination
        existing = client.destinations.list(name=payload["name"], per_page=1)
        if existing:
            destination = client.destinations.update(existing[0].id, payload)
            print(f"✅ Updated existing destination: {destination.name}")
            return destination
        else:
            raise Exception("Resource conflict but destination not found")

Error Context and Debugging​

Exception Attributes​

All SDK exceptions provide rich context:

Exception Context
try:
    flow = client.flows.get("invalid-id")
except NexlaAPIError as e:
    print(f"Operation: {e.operation}")
    print(f"Resource Type: {e.resource_type}")
    print(f"Resource ID: {e.resource_id}")
    print(f"Status Code: {e.status_code}")
    print(f"Message: {e.message}")
    print(f"Retry After: {e.retry_after}")

Logging Errors​

Error Logging
import logging

logger = logging.getLogger(__name__)

try:
    flows = client.flows.list()
except NexlaAPIError as e:
    logger.error(
        f"API Error: {e.message}",
        extra={
            "operation": e.operation,
            "resource_type": e.resource_type,
            "resource_id": e.resource_id,
            "status_code": e.status_code
        }
    )
    raise

Best Practices​

1. Always Handle Authentication Errors​

Auth Error Handling
try:
    flows = client.flows.list()
except AuthenticationError:
    # Don't retry auth errors - they won't succeed
    logger.error("Authentication failed - check credentials")
    sys.exit(1)

2. Implement Exponential Backoff for Rate Limits​

Rate Limit Backoff
import time
from nexla_sdk.exceptions import RateLimitError

def with_backoff(func, max_retries=5):
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError as e:
            wait_time = e.retry_after or (2 ** attempt)
            if attempt < max_retries - 1:
                time.sleep(wait_time)
                continue
            raise

3. Log All Errors with Context​

Comprehensive Logging
def log_error_with_context(e, operation="unknown"):
    logger.error(
        f"Error in {operation}: {str(e)}",
        extra={
            "error_type": type(e).__name__,
            "operation": operation,
            "traceback": traceback.format_exc()
        }
    )

Next Steps​

Learn More

For detailed error handling patterns and retry strategies, see the Error Handling & Retries page in our Nexla SDK guides.