Exceptions

This module defines the exception hierarchy for litestar-storages. All storage backends raise exceptions from this hierarchy, enabling consistent error handling across different providers.

Note

Exception names are prefixed with Storage to avoid shadowing Python built-in exceptions like FileNotFoundError and PermissionError.

Exception Hierarchy

Exception
+-- StorageError (base for all storage exceptions)
    +-- StorageFileNotFoundError
    +-- StorageFileExistsError
    +-- StoragePermissionError
    +-- StorageConnectionError
    +-- ConfigurationError

Base Exception

class litestar_storages.exceptions.StorageError

Bases: Exception

Base exception for all storage-related errors.

All storage backends should raise exceptions derived from this class to allow for consistent error handling across different storage providers.

File Exceptions

class litestar_storages.exceptions.StorageFileNotFoundError

Bases: StorageError

Raised when a requested file does not exist in storage.

Variables:

key – The storage path/key that was not found

Parameters:

key (str)

Example

from litestar_storages.exceptions import StorageFileNotFoundError

try:
    data = await storage.get_bytes("nonexistent.txt")
except StorageFileNotFoundError as e:
    print(f"File not found: {e.key}")

__init__(key)

Initialize StorageFileNotFoundError.

Parameters:

key (str) – The storage path/key that was not found

class litestar_storages.exceptions.StorageFileExistsError

Bases: StorageError

Raised when attempting to create a file that already exists (when overwrite is disabled).

Variables:

key – The storage path/key that already exists

Parameters:

key (str)

__init__(key)

Initialize StorageFileExistsError.

Parameters:

key (str) – The storage path/key that already exists

Permission and Connection Exceptions

class litestar_storages.exceptions.StoragePermissionError

Bases: StorageError

Raised when the operation fails due to insufficient permissions.

This typically occurs when: - The storage backend credentials lack necessary permissions - File system permissions prevent the operation - Access control policies block the operation

Common causes:

• Insufficient IAM permissions (cloud backends)

• File system permissions blocking access

• Access control policies (bucket policies, ACLs)

class litestar_storages.exceptions.StorageConnectionError

Bases: StorageError

Raised when unable to connect to the storage backend.

This typically occurs when: - Network connectivity issues prevent access - Storage service is unavailable - Authentication fails - Invalid endpoint configuration

Common causes:

• Network connectivity issues

• Storage service unavailable

• Authentication failures

• Invalid endpoint configuration

Configuration Exception

class litestar_storages.exceptions.ConfigurationError

Bases: StorageError

Raised when storage backend configuration is invalid.

This typically occurs when: - Required configuration parameters are missing - Configuration values are invalid or incompatible - Environment variables are malformed

Common causes:

• Missing required parameters (bucket name, credentials)

• Invalid configuration values

• Missing optional dependencies (aioboto3, aiofiles, etc.)

Error Handling Example

Comprehensive error handling pattern:

from litestar_storages import Storage, StorageError
from litestar_storages.exceptions import (
    ConfigurationError,
    StorageConnectionError,
    StorageFileNotFoundError,
    StoragePermissionError,
)


async def safe_get_file(storage: Storage, key: str) -> bytes | None:
    """Safely retrieve a file with comprehensive error handling."""
    try:
        return await storage.get_bytes(key)

    except StorageFileNotFoundError:
        # File doesn't exist - handle gracefully
        return None

    except StoragePermissionError:
        # Log and re-raise or handle based on use case
        logger.error(f"Permission denied for key: {key}")
        raise

    except StorageConnectionError as e:
        # Transient error - could retry
        logger.warning(f"Connection error: {e}, retrying...")
        raise

    except ConfigurationError as e:
        # Configuration issue - fatal
        logger.critical(f"Storage misconfigured: {e}")
        raise

    except StorageError as e:
        # Catch-all for other storage errors
        logger.error(f"Storage error: {e}")
        raise


async def upload_with_retry(
    storage: Storage,
    key: str,
    data: bytes,
    max_retries: int = 3,
) -> StoredFile:
    """Upload with automatic retry on transient errors."""
    for attempt in range(max_retries):
        try:
            return await storage.put(key, data)
        except StorageConnectionError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff
    raise RuntimeError("Unreachable")