Error Handling

Handle errors gracefully with the SDK's comprehensive exception hierarchy.

What This Section Covers

Exception types, error handling patterns, retry behavior, and best practices for robust error handling in production applications using both sync and async clients.

Exception Hierarchy

All SDK exceptions inherit from CustodyError:

CustodyError (base exception)
├── AuthenticationError           # 401 - Invalid JWT token
├── AuthorizationError            # 403 - Insufficient permissions
├── ValidationError               # 400 - Invalid request parameters
├── NotFoundError                 # 404 - Resource doesn't exist
│   ├── AuthorizationNotFoundError
│   └── TransferNotFoundError
├── ConflictError                 # 409 - Resource state conflict
│   ├── DuplicateAuthorizationError
│   ├── PoolConflictError
│   ├── MaxAuthorizationsExceededError
│   ├── InvalidTransferSequenceError
│   └── SessionAlreadyClosedError
├── RateLimitError                # 429 - Rate limit exceeded
├── ServiceUnavailableError       # 503 - Service temporarily unavailable
├── StorageError                  # 500 - Storage backend error
└── InternalError                 # 500 - Internal server error

Basic Error Handling

Catching Specific Exceptions

from vecu_custody import CustodyClient
from vecu_custody.exceptions import (
    AuthorizationNotFoundError,
    AuthenticationError,
    CustodyError
)

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.get(authorization_id="AUTH-12345678")
except AuthorizationNotFoundError as e:
    print(f"Authorization not found: {e.authorization_id}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except CustodyError as e:
    print(f"API error: {e}")

Exception Attributes

All exceptions provide detailed information:

from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    client.authorizations.create(
        vin="INVALID",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_test123",
        authorized_by="system-integration",
        make_model="Honda Accord 2023"
    )
except CustodyError as e:
    print(f"Error: {e}")
    print(f"HTTP Status: {e.status_code}")
    print(f"Correlation ID: {e.correlation_id}")

Common Error Scenarios

Authentication Errors

from vecu_custody import CustodyClient
from vecu_custody.exceptions import AuthenticationError

try:
    client = CustodyClient.sandbox(token="invalid-token")
    client.authorizations.list(vin="9HGBH41JXMN999999")
except AuthenticationError as e:
    print("Invalid JWT token - check your credentials")
    # Re-authenticate or alert administrators

Validation Errors

from vecu_custody import CustodyClient
from vecu_custody.exceptions import ValidationError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    client.authorizations.create(
        vin="",  # Empty VIN - invalid
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_test123",
        authorized_by="system-integration",
        make_model="Honda Accord 2023"
    )
except ValidationError as e:
    print(f"Validation failed: {e}")

Resource Not Found

from vecu_custody import CustodyClient
from vecu_custody.exceptions import AuthorizationNotFoundError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.get(authorization_id="AUTH-NONEXISTENT")
except AuthorizationNotFoundError as e:
    print(f"Authorization {e.authorization_id} does not exist")
    # Handle gracefully

Duplicate Authorization

from vecu_custody import CustodyClient
from vecu_custody.exceptions import DuplicateAuthorizationError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    client.authorizations.create(
        vin="9HGBH41JXMN999999",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_test123",
        authorized_by="system-integration",
        make_model="Honda Accord 2023"
    )
except DuplicateAuthorizationError as e:
    print(f"Authorization already exists: {e.existing_authorization_id}")

Pool Conflict

from vecu_custody import CustodyClient
from vecu_custody.exceptions import PoolConflictError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    client.authorizations.create(...)
except PoolConflictError as e:
    print(f"Route owned by different client: {e.existing_client_id}")

Rate Limiting

import time
from vecu_custody import CustodyClient
from vecu_custody.exceptions import RateLimitError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auths = client.authorizations.list(vin="9HGBH41JXMN999999")
except RateLimitError as e:
    retry_after = e.retry_after  # Seconds to wait
    print(f"Rate limited. Retry after {retry_after} seconds")
    time.sleep(retry_after)
    # Retry request
    auths = client.authorizations.list(vin="9HGBH41JXMN999999")

Service Unavailable

from vecu_custody import CustodyClient
from vecu_custody.exceptions import ServiceUnavailableError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(...)
except ServiceUnavailableError as e:
    print("Service temporarily unavailable - retry with backoff")

Error Handling Patterns

Pattern 1: Try-Except-Else

from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(
        vin="9HGBH41JXMN999999",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_test123",
        authorized_by="system-integration",
        make_model="Honda Accord 2023"
    )
except CustodyError as e:
    print(f"Failed to create authorization: {e}")
else:
    # Only runs if no exception
    print(f"Authorization created: {auth.authorization_id}")

Pattern 2: Multiple Exception Types

import time
from vecu_custody import CustodyClient
from vecu_custody.exceptions import (
    AuthenticationError,
    ValidationError,
    RateLimitError,
    ServiceUnavailableError,
    CustodyError
)

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(...)
except AuthenticationError:
    # Re-authenticate
    print("Authentication failed - check credentials")
except ValidationError as e:
    # Fix request parameters
    print(f"Invalid request: {e}")
except RateLimitError as e:
    # Wait and retry
    time.sleep(e.retry_after)
except ServiceUnavailableError:
    # Service issue - retry with backoff
    print("Service unavailable - retrying...")
except CustodyError as e:
    # Catch-all for other errors
    print(f"Unexpected error: {e}")

Pattern 3: Context Manager with Error Handling

from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError

def process_transfers():
    with CustodyClient.sandbox(token="your-jwt-token") as client:
        try:
            for transfer in client.transfers.get_history(vin="9HGBH41JXMN999999"):
                # Process transfer
                print(f"Transfer: {transfer.transfer_type}")
        except CustodyError as e:
            print(f"Error processing transfers: {e}")
            raise  # Re-raise for caller to handle

Pattern 4: Custom Error Handler with Retry

import time
from functools import wraps
from vecu_custody import CustodyClient
from vecu_custody.exceptions import (
    CustodyError,
    RateLimitError,
    ServiceUnavailableError,
)

def retry_on_error(max_retries=3, backoff=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except RateLimitError as e:
                    if attempt < max_retries - 1:
                        time.sleep(e.retry_after)
                        continue
                    raise
                except ServiceUnavailableError:
                    if attempt < max_retries - 1:
                        wait = backoff ** attempt
                        print(f"Retry {attempt + 1}/{max_retries} after {wait}s")
                        time.sleep(wait)
                        continue
                    raise
                except CustodyError:
                    # Don't retry other errors
                    raise
        return wrapper
    return decorator

@retry_on_error(max_retries=3)
def create_authorization(client: CustodyClient, vin: str):
    return client.authorizations.create(
        vin=vin,
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_test123",
        authorized_by="system-integration",
        make_model="Honda Accord 2023"
    )

Automatic Retry Behavior

The SDK automatically retries certain errors with exponential backoff.

Retry Configuration

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(
    token="your-jwt-token",
    max_retries=5,  # Retry up to 5 times (default: 3)
    timeout=60      # Timeout in seconds (default: 30)
)

Disable Retries

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(
    token="your-jwt-token",
    max_retries=0  # No retries
)

Logging Errors

Basic Logging

import logging
from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError

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

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(...)
except CustodyError as e:
    logger.error(
        "Failed to create authorization",
        extra={
            "error": str(e),
            "correlation_id": getattr(e, "correlation_id", None),
            "status_code": getattr(e, "status_code", None)
        }
    )

Structured Logging

import structlog
from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError

logger = structlog.get_logger()
client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(...)
except CustodyError as e:
    logger.error(
        "authorization_creation_failed",
        error=str(e),
        correlation_id=getattr(e, "correlation_id", None),
        status_code=getattr(e, "status_code", None)
    )

Best Practices

1. Catch Specific Exceptions First

from vecu_custody.exceptions import AuthorizationNotFoundError, CustodyError

# Good
try:
    auth = client.authorizations.get(authorization_id="...")
except AuthorizationNotFoundError:
    # Handle not found
    pass
except CustodyError:
    # Handle other errors
    pass

# Avoid
try:
    auth = client.authorizations.get(authorization_id="...")
except CustodyError:
    # Can't distinguish error types
    pass

2. Log Correlation IDs

Always log correlation IDs for debugging with support:

from vecu_custody.exceptions import CustodyError

try:
    auth = client.authorizations.create(...)
except CustodyError as e:
    logger.error(f"API error (correlation_id: {e.correlation_id}): {e}")

3. Don't Silence Errors

from vecu_custody.exceptions import CustodyError

# Avoid
try:
    auth = client.authorizations.create(...)
except CustodyError:
    pass  # Silent failure - bad!

# Good
try:
    auth = client.authorizations.create(...)
except CustodyError as e:
    logger.error(f"Failed to create authorization: {e}")
    raise  # Re-raise or handle appropriately

4. Use Type Hints

from vecu_custody import CustodyClient
from vecu_custody.exceptions import CustodyError
from vecu_custody.models import Authorization

def create_authorization(client: CustodyClient, vin: str) -> Authorization | None:
    try:
        return client.authorizations.create(
            vin=vin,
            origin="LOC-AUCTION-MANHEIM-ATLANTA",
            destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
            person_identity_key="vecu_test123",
            authorized_by="system-integration",
            make_model="Honda Accord 2023"
        )
    except CustodyError as e:
        logger.error(f"Failed to create authorization: {e}")
        return None

Complete Exception Reference

from vecu_custody.exceptions import (
    # Base exception
    CustodyError,

    # Authentication/Authorization
    AuthenticationError,      # 401 - Invalid token
    AuthorizationError,       # 403 - Insufficient permissions

    # Validation
    ValidationError,          # 400 - Invalid parameters

    # Not Found
    NotFoundError,            # 404 - Base not found
    AuthorizationNotFoundError,
    TransferNotFoundError,

    # Conflicts
    ConflictError,            # 409 - Base conflict
    DuplicateAuthorizationError,
    PoolConflictError,
    MaxAuthorizationsExceededError,
    InvalidTransferSequenceError,
    SessionAlreadyClosedError,

    # Server Errors
    RateLimitError,           # 429
    ServiceUnavailableError,  # 503
    StorageError,             # 500
    InternalError,            # 500
)

Next Steps

Testing - Test error handling
Advanced Async Patterns - Async error handling with concurrency
API Reference - Complete API documentation

Vehicle Custody