Documentation
SDK Reference: Exceptions
Exception hierarchy and error types in the SDK
Exception Handling
All SDK exceptions inherit from AionvisionError and are available as class attributes on AionVision, so you only need a single import: from aion import AionVision. Then use AionVision.AuthenticationError, etc.
Exception Hierarchy
AionvisionError (base)├── AuthenticationError # Invalid API key (401)├── AionvisionPermissionError # Insufficient permissions (403)├── ResourceNotFoundError # Resource not found (404)├── ValidationError # Invalid request data (400/422)├── QuotaExceededError # Upload/API quota exceeded├── RateLimitError # Too many requests (429)├── AionvisionTimeoutError # Operation timed out├── ServerError # Server-side error (5xx)├── AionvisionConnectionError # Network connection failed├── CircuitBreakerError # Circuit breaker open├── UploadError # Upload operation failed├── DescriptionError # Description generation failed├── ChatError # Chat operation failed├── BatchError # Batch operation failed├── DocumentProcessingError # Document processing failed├── CloudStorageError # Cloud storage job failed└── SSEStreamError # SSE streaming failedBase Exception
AionvisionError
Base exception for all SDK errors
class AionvisionError(Exception): code: str # Error code (e.g., "AUTH_ERROR") message: str # Human-readable message details: dict[str, Any] # Additional context
def __init__( self, code: str, message: str, *, details: Optional[dict[str, Any]] = None, ) -> None: ...Example
from aion import AionVision
try: result = await client.upload("image.jpg")except AionVision.AionvisionError as e: print(f"Error code: {e.code}") print(f"Message: {e.message}") print(f"Details: {e.details}")Authentication Errors
AuthenticationError
Invalid or missing API key (HTTP 401)
class AuthenticationError(AionvisionError): # code = "AUTH_ERROR" def __init__(self, message: str = "Authentication failed") -> None: ...Example
from aion import AionVision
try: result = await client.upload("image.jpg")except AionVision.AuthenticationError: print("Invalid API key - check your credentials")AionvisionPermissionError
Insufficient permissions (HTTP 403)
class AionvisionPermissionError(AionvisionError): # code = "PERMISSION_DENIED" def __init__(self, message: str = "Permission denied") -> None: ...Rate Limiting Errors
RateLimitError
API rate limit exceeded (HTTP 429)
class RateLimitError(AionvisionError): # code = "RATE_LIMIT" retry_after: Optional[int] # Seconds to wait before retrying limit: Optional[int] # Max requests allowed remaining: Optional[int] # Requests remaining (always 0) reset: Optional[int] # Unix timestamp when limit resets
def __init__( self, message: str = "Rate limit exceeded", *, retry_after: Optional[int] = None, limit: Optional[int] = None, remaining: Optional[int] = None, reset: Optional[int] = None, ) -> None: ...Example
import asynciofrom aion import AionVision
try: result = await client.upload("image.jpg")except AionVision.RateLimitError as e: print(f"Rate limited: {e.message}") print(f"Retry after: {e.retry_after} seconds") await asyncio.sleep(e.retry_after or 5) result = await client.upload("image.jpg")QuotaExceededError
Upload or processing quota exceeded
class QuotaExceededError(AionvisionError): # code = "QUOTA_EXCEEDED" # details contains: available, limit
def __init__( self, message: str, *, available: int = 0, limit: int = 0, partial_results: Any = None, ) -> None: ...Example
from aion import AionVision
try: result = await client.upload("image.jpg")except AionVision.QuotaExceededError as e: print(f"Quota exceeded: {e.message}") print(f"Available: {e.details.get('available')}") print(f"Limit: {e.details.get('limit')}")Validation Errors
ValidationError
Request validation failed (HTTP 400/422)
class ValidationError(AionvisionError): # code = "VALIDATION_ERROR"
def __init__( self, message: str, *, details: Optional[dict[str, Any]] = None, ) -> None: ...ResourceNotFoundError
Requested resource not found (HTTP 404)
class ResourceNotFoundError(AionvisionError): # code = "NOT_FOUND" # details contains: resource_type, resource_id
def __init__( self, resource_type: str, resource_id: str, *, message: Optional[str] = None, details: Optional[dict[str, Any]] = None, ) -> None: ...Example
from aion import AionVision
try: details = await client.files.get(file_id="nonexistent")except AionVision.ResourceNotFoundError as e: print(f"Not found: {e.message}") print(f"Resource type: {e.details.get('resource_type')}") print(f"Resource ID: {e.details.get('resource_id')}")Network Errors
AionvisionTimeoutError
Operation timed out
class AionvisionTimeoutError(AionvisionError): # code = "TIMEOUT" last_result: Optional[Any] # Last result before timeout
def __init__( self, message: str = "Operation timed out", *, last_result: Optional[Any] = None, ) -> None: ...AionvisionConnectionError
Network connection failed
class AionvisionConnectionError(AionvisionError): # code = "CONNECTION_ERROR" def __init__(self, message: str = "Connection failed") -> None: ...ServerError
Server-side error (HTTP 5xx)
class ServerError(AionvisionError): # code = "SERVER_ERROR" def __init__(self, message: str = "Server error") -> None: ...CircuitBreakerError
Circuit breaker open due to repeated failures
class CircuitBreakerError(AionvisionError): # code = "CIRCUIT_OPEN" endpoint: str # Endpoint that triggered the circuit retry_after: float # Seconds until circuit allows requests
def __init__(self, endpoint: str, retry_after: float) -> None: ...Operation Errors
UploadError
File upload operation failed
class UploadError(AionvisionError): # code = "UPLOAD_ERROR" # details contains: filename, stage
def __init__( self, message: str, *, filename: Optional[str] = None, stage: Optional[str] = None, # quota | presigned | upload | confirm | session_results | chunk_upload | prepare | expand session_id: Optional[str] = None, partial_results: Any = None, ) -> None: ...Example
from aion import AionVision
try: result = await client.upload("image.jpg")except AionVision.UploadError as e: print(f"Upload failed: {e.message}") print(f"Filename: {e.details.get('filename')}") print(f"Stage: {e.details.get('stage')}")DescriptionError
Description generation failed
class DescriptionError(AionvisionError): # code = "DESCRIPTION_ERROR" # details contains: image_id, status
def __init__( self, message: str, *, image_id: Optional[str] = None, status: Optional[str] = None, ) -> None: ...Note
By default, upload() raises DescriptionError for description failures (raise_on_failure=True). Set raise_on_failure=False to handle failures manually.
DocumentProcessingError
Document processing operation failed
class DocumentProcessingError(AionvisionError): # code = "DOCUMENT_PROCESSING_ERROR" # details contains: document_id, status
def __init__( self, message: str, *, document_id: Optional[str] = None, status: Optional[str] = None, ) -> None: ...Example
from aion import AionVision
try: await client.documents.delete(document_id="doc_abc123")except AionVision.DocumentProcessingError as e: print(f"Cannot process document: {e.message}") print(f"Document ID: {e.details.get('document_id')}")ChatError
Chat operation failed
class ChatError(AionvisionError): # code = "CHAT_ERROR" or custom error_code # details contains: session_id
def __init__( self, message: str, *, session_id: Optional[str] = None, error_code: Optional[str] = None, ) -> None: ...BatchError
Batch operation failed
class BatchError(AionvisionError): # code = "BATCH_ERROR" # details contains: batch_id, failed_items
def __init__( self, message: str, *, batch_id: Optional[str] = None, failed_items: Optional[list[str]] = None, ) -> None: ...CloudStorageError
Cloud storage import/export job failed
class CloudStorageError(AionvisionError): # code = "CLOUD_STORAGE_ERROR" # details contains: job_id, job_status
def __init__( self, message: str, *, job_id: Optional[str] = None, job_status: Optional[str] = None, ) -> None: ...Example
from aion import AionVision
try: job = await client.cloud_storage.import_and_wait( connection_id="conn_abc123", files=files, )except AionVision.CloudStorageError as e: print(f"Job failed: {e.message}") print(f"Job ID: {e.details.get('job_id')}") print(f"Status: {e.details.get('job_status')}")SSEStreamError
SSE streaming connection failed after reconnection attempts
class SSEStreamError(AionvisionError): # code = "SSE_STREAM_ERROR" last_event_id: Optional[str] # Last successfully received event ID events_received: int # Total events received before failure reconnect_attempts: int # Number of reconnection attempts made
def __init__( self, message: str, *, last_event_id: Optional[str] = None, events_received: int = 0, reconnect_attempts: int = 0, ) -> None: ...Error Handling Patterns
Comprehensive Error Handling
from aion import AionVision
async def safe_upload(client: AionVision, file): try: return await client.upload(file)
except AionVision.AuthenticationError: print("Invalid API key") raise
except AionVision.RateLimitError as e: print(f"Rate limited. Retry after {e.retry_after}s") raise
except AionVision.QuotaExceededError as e: print(f"Quota exceeded: {e.details.get('available')}/{e.details.get('limit')}") raise
except AionVision.ValidationError as e: print(f"Invalid request: {e.message}") print(f"Details: {e.details}") raise
except AionVision.AionvisionError as e: print(f"API error [{e.code}]: {e.message}") raiseRetry with Exponential Backoff
import asynciofrom aion import AionVision
async def upload_with_retry(client: AionVision, file, max_attempts=3): for attempt in range(max_attempts): try: return await client.upload(file)
except AionVision.RateLimitError as e: if attempt < max_attempts - 1: wait = e.retry_after or (2 ** attempt) print(f"Rate limited, waiting {wait}s...") await asyncio.sleep(wait) else: raise
except (AionVision.ServerError, AionVision.AionvisionConnectionError): if attempt < max_attempts - 1: wait = 2 ** attempt print(f"Transient error, retrying in {wait}s...") await asyncio.sleep(wait) else: raise