Base Classes

This module defines the core storage protocol and abstract base class that all storage backends implement. The protocol-based design ensures consistent behavior across different storage providers while allowing for backend-specific optimizations.

Storage Protocol

The Storage protocol defines the interface that all storage backends must implement. It is marked with @runtime_checkable, allowing runtime type checking with isinstance().

class litestar_storages.base.Storage

Bases: Protocol

Async storage protocol.

All storage backends must implement this protocol to ensure consistent behavior across different storage providers.

This protocol defines the core interface for async file storage operations, including uploading, downloading, deleting, and managing file metadata.

Core Operations

The protocol defines these core async operations:

• put() - Store data at a key

• get() - Retrieve data as an async stream

• get_bytes() - Retrieve data as bytes

• delete() - Remove a file

• exists() - Check if a file exists

• list() - List files with optional prefix filter

• url() - Generate access URL (presigned for cloud backends)

• copy() - Copy a file within the storage

• move() - Move/rename a file

• info() - Get file metadata without downloading

• close() - Release resources

async put(key, data, *, content_type=None, metadata=None)

Store data at the given key.

Parameters:

• key (str) – Storage path/key for the file. This should be a forward-slash separated path (e.g., “images/photo.jpg”).

• data (bytes | AsyncGenerator[bytes, None]) – File contents as bytes or async byte stream. For large files, prefer using an async iterator to avoid loading the entire file into memory.

• content_type (str | None) – MIME type of the content (e.g., ‘image/jpeg’, ‘text/plain’). If not provided, backends may attempt to infer it from the file extension.

• metadata (dict[str, str] | None) – Additional metadata to store with the file. Keys and values must be strings. Backend-specific limits may apply.

Return type:

StoredFile

Returns:

StoredFile containing metadata about the stored file, including the actual size, etag, and last modified timestamp.

Raises:

• StorageError – If the upload fails for any reason

• StoragePermissionError – If lacking permissions to write to this location

• ConfigurationError – If storage backend is misconfigured

get(key)

Retrieve file contents as an async byte stream.

This method returns an async iterator that yields chunks of the file, allowing for efficient streaming of large files without loading the entire content into memory.

Parameters:

key (str) – Storage path/key for the file

Yields:

Chunks of file data as bytes. Chunk size is backend-dependent.

Raises:

• StorageFileNotFoundError – If the file does not exist

• StorageError – If the retrieval fails

• StoragePermissionError – If lacking permissions to read this file

Return type:

AsyncIterator[bytes]

async get_bytes(key)

Retrieve entire file contents as bytes.

This is a convenience method that collects the stream from get() into memory. Use get() for large files to avoid memory pressure.

Parameters:

key (str) – Storage path/key for the file

Return type:

bytes

Returns:

Complete file contents as bytes

Raises:

• StorageFileNotFoundError – If the file does not exist

• StorageError – If the retrieval fails

• StoragePermissionError – If lacking permissions to read this file

async delete(key)

Delete a file.

Parameters:

key (str) – Storage path/key for the file

Raises:

• StorageFileNotFoundError – If the file does not exist (implementation-dependent; some backends may silently succeed for idempotency)

• StorageError – If the deletion fails

• StoragePermissionError – If lacking permissions to delete this file

Return type:

None

async exists(key)

Check if a file exists.

Parameters:

key (str) – Storage path/key for the file

Return type:

bool

Returns:

True if the file exists, False otherwise

Raises:

StorageError – If the check fails (rare; typically returns False)

async list(prefix='', *, limit=None)

List files with optional prefix filter.

Parameters:

• prefix (str) – Filter results to keys starting with this prefix. Use empty string to list all files. For hierarchical storage, use forward slashes (e.g., “images/2024/”).

• limit (int | None) – Maximum number of results to return. If None, returns all matching files. Note: backends may have their own internal limits.

Yields:

StoredFile metadata for each matching file, typically in lexicographical order by key.

Raises:

• StorageError – If the listing operation fails

• StoragePermissionError – If lacking permissions to list files

Return type:

AsyncGenerator[StoredFile, None]

async url(key, *, expires_in=None)

Generate a URL for accessing the file.

For cloud backends, this typically generates a presigned URL that grants temporary access to the file. For filesystem backends, this returns a path or configured base URL.

Parameters:

• key (str) – Storage path/key for the file

• expires_in (timedelta | None) – Optional expiration time for signed URLs. If None, uses backend’s default expiration (typically 1 hour). For filesystem backends, this parameter may be ignored.

Returns:

• A presigned URL with embedded credentials (cloud backends)

• A public URL if the file is publicly accessible

• A file:// URL or relative path (filesystem backend)

Return type:

URL string for accessing the file. This may be

Raises:

• StorageFileNotFoundError – If the file does not exist (implementation-dependent)

• StorageError – If URL generation fails

• StoragePermissionError – If lacking permissions to generate URLs

Note

URLs may become invalid after the expiration time or if the file is deleted. Clients should handle 404/403 responses gracefully.

async copy(source, destination)

Copy a file within the storage backend.

This operation should be atomic where possible. Some backends can perform server-side copies without downloading and re-uploading data.

Parameters:

• source (str) – Source key to copy from

• destination (str) – Destination key to copy to. If this key already exists, it will be overwritten.

Return type:

StoredFile

Returns:

StoredFile metadata for the new copy at the destination

Raises:

• StorageFileNotFoundError – If the source file does not exist

• StorageError – If the copy operation fails

• StoragePermissionError – If lacking permissions for the operation

async move(source, destination)

Move/rename a file within the storage backend.

This operation should be atomic where possible. Default implementation performs a copy followed by delete.

Parameters:

• source (str) – Source key to move from

• destination (str) – Destination key to move to. If this key already exists, it will be overwritten.

Return type:

StoredFile

Returns:

StoredFile metadata for the file at the new destination

Raises:

• StorageFileNotFoundError – If the source file does not exist

• StorageError – If the move operation fails

• StoragePermissionError – If lacking permissions for the operation

async info(key)

Get metadata about a file without downloading it.

This is useful for checking file size, content type, and other metadata without incurring the cost of downloading the file.

Parameters:

key (str) – Storage path/key for the file

Return type:

StoredFile

Returns:

StoredFile with metadata including size, content_type, etag, last_modified, and any custom metadata

Raises:

• StorageFileNotFoundError – If the file does not exist

• StorageError – If the metadata retrieval fails

• StoragePermissionError – If lacking permissions to access file metadata

async close()

Close the storage backend and release resources.

This method should be called when the storage is no longer needed, typically during application shutdown. It allows backends to clean up resources like HTTP sessions, connection pools, or file handles.

For backends that don’t require cleanup, this method is a no-op.

Note

After calling close(), the storage instance should not be used. Some backends may raise errors if operations are attempted after close().

Return type:

None

__init__(*args, **kwargs)

Base Storage Class

The BaseStorage abstract base class provides default implementations for convenience methods while requiring subclasses to implement core operations.

class litestar_storages.base.BaseStorage

Bases: ABC

Abstract base class providing common functionality for storage backends.

Backends can inherit from this to get default implementations of convenience methods while only implementing core abstract operations.

Subclasses must implement: - put() - get() - delete() - exists() - list() - url() - info()

This class provides default implementations for: - get_bytes() - collects stream into memory - copy() - downloads and re-uploads - move() - copies then deletes

Abstract Methods (must implement)

Subclasses must implement these methods:

• put()

• get()

• delete()

• exists()

• list()

• url()

• info()

Default Implementations

These methods have default implementations that can be overridden for optimization:

• get_bytes() - Collects stream into bytes

• copy() - Downloads and re-uploads (override for server-side copy)

• move() - Copies then deletes (override for atomic rename)

• close() - No-op (override to release resources)

abstractmethod async put(key, data, *, content_type=None, metadata=None)

Store data at the given key. Must be implemented by subclasses.

Parameters:

• key (str)

• data (bytes | AsyncGenerator[bytes, None])

• content_type (str | None)

• metadata (dict[str, str] | None)

Return type:

StoredFile

abstractmethod get(key)

Retrieve file as async byte stream. Must be implemented by subclasses.

Parameters:

key (str)

Return type:

AsyncIterator[bytes]

abstractmethod async delete(key)

Delete a file. Must be implemented by subclasses.

Parameters:

key (str)

Return type:

None

abstractmethod async exists(key)

Check if file exists. Must be implemented by subclasses.

Parameters:

key (str)

Return type:

bool

abstractmethod async list(prefix='', *, limit=None)

List files with prefix filter. Must be implemented by subclasses.

Parameters:

• prefix (str)

• limit (int | None)

Return type:

AsyncGenerator[StoredFile, None]

abstractmethod async url(key, *, expires_in=None)

Generate URL for file access. Must be implemented by subclasses.

Parameters:

• key (str)

• expires_in (timedelta | None)

Return type:

str

abstractmethod async info(key)

Get file metadata. Must be implemented by subclasses.

Parameters:

key (str)

Return type:

StoredFile

async get_bytes(key)

Default implementation: collect stream into bytes.

This method gathers all chunks from get() and joins them into a single bytes object. For large files, prefer using get() directly to stream the content.

Parameters:

key (str) – Storage path/key for the file

Return type:

bytes

Returns:

Complete file contents as bytes

Raises:

• StorageFileNotFoundError – If the file does not exist

• StorageError – If the retrieval fails

async copy(source, destination)

Default implementation: download and re-upload.

This implementation downloads the source file completely into memory, then uploads it to the destination. Backends that support server-side copy should override this method for better performance.

Parameters:

• source (str) – Source key to copy from

• destination (str) – Destination key to copy to

Return type:

StoredFile

Returns:

StoredFile metadata for the new copy

Raises:

• StorageFileNotFoundError – If source does not exist

• StorageError – If the operation fails

async move(source, destination)

Default implementation: copy then delete.

This implementation performs a copy operation followed by deletion of the source. Backends that support atomic move/rename should override this method.

Parameters:

• source (str) – Source key to move from

• destination (str) – Destination key to move to

Return type:

StoredFile

Returns:

StoredFile metadata for the moved file

Raises:

• StorageFileNotFoundError – If source does not exist

• StorageError – If the operation fails

Note

This operation is not atomic in the default implementation. If deletion fails after a successful copy, the source file will remain.

async close()

Default implementation: no-op.

Subclasses that manage resources (HTTP sessions, connection pools, etc.) should override this method to properly release them.

Return type:

None

Usage Example

Implementing a custom storage backend:

from collections.abc import AsyncGenerator, AsyncIterator
from datetime import timedelta

from litestar_storages import BaseStorage, StoredFile


class MyCustomStorage(BaseStorage):
    """Custom storage backend example."""

    def __init__(self, connection_url: str) -> None:
        self.connection_url = connection_url
        self._client = None

    async def put(
        self,
        key: str,
        data: bytes | AsyncGenerator[bytes, None],
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
    ) -> StoredFile:
        # Implementation here
        ...

    async def get(self, key: str) -> AsyncIterator[bytes]:
        # Implementation here
        ...

    async def delete(self, key: str) -> None:
        # Implementation here
        ...

    async def exists(self, key: str) -> bool:
        # Implementation here
        ...

    async def list(
        self,
        prefix: str = "",
        *,
        limit: int | None = None,
    ) -> AsyncGenerator[StoredFile, None]:
        # Implementation here
        ...

    async def url(
        self,
        key: str,
        *,
        expires_in: timedelta | None = None,
    ) -> str:
        # Implementation here
        ...

    async def info(self, key: str) -> StoredFile:
        # Implementation here
        ...

    async def close(self) -> None:
        # Clean up resources
        if self._client:
            await self._client.close()