API Reference

Protocols

BlobStoreProtocol

source

Protocol for blob storage operations.

This interface defines the contract for file storage backends (S3, GCS, Azure Blob, local filesystem, etc.).

Example

class S3BlobStore:
    async def upload(self, path: str, data: bytes, **options) -> FileInfo:
        await self._client.put_object(Bucket=self._bucket, Key=path, Body=data)
        return FileInfo(path=path, size=len(data), ...)

upload

async def upload(
    path: str,
    data: bytes | AsyncIterator[bytes],
    content_type: str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to the storage backend.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`data`	bytes | AsyncIterator[bytes]	File content as bytes or async iterator.
`content_type`	str | None	MIME type of the content. **options: Additional upload options.

Returns

Type	Description
FileInfo	FileInfo with path, size, and metadata.

download

async def download(path: str) -> bytes

source

Download file content into memory.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
bytes	File content as bytes.

stream

def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncIterator[bytes]

source

Stream file content (memory efficient).

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`chunk_size`	int	Size of each chunk in bytes.

Yields

Type	Description
AsyncIterator[bytes]	File content in chunks.

delete

async def delete(path: str) -> None

source

Delete a file.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

exists

async def exists(path: str) -> bool

source

Check if file exists.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
bool	True if file exists.

info

async def info(path: str) -> FileInfo

source

Get file metadata.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
FileInfo	FileInfo with size, content_type, etc.

list

def list(prefix: str = '') -> AsyncIterator[FileInfo]

source

List files with a given prefix.

Parameters

Parameter	Type	Description
`prefix`	str	Path prefix to filter by.

Yields

Type	Description
AsyncIterator[FileInfo]	FileInfo for each matching file.

get_url

async def get_url(path: str) -> str

source

Get public URL (if applicable).

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
str	Public URL string.

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Get a temporary secure URL.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`expires_in`	timedelta	URL validity window as a timedelta (default one hour). Pass ``timedelta(minutes=5)`` for secure short-lived downloads or ``timedelta(hours=24)`` for bulk exports.
`method`	str	HTTP method (GET or PUT).

Returns

Type	Description
str	Presigned URL string.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

---

Classes

AzureDriver

source

Azure Blob Storage driver.

Wraps azure-storage-blob (async variant) to provide fully async Blob Storage operations.

Install the optional dependency with

pip install lexigram-storage[azure]
# i.e.  azure-storage-blob>=12.20.0

pip install lexigram-storage[azure]
# i.e.  azure-storage-blob>=12.20.0

Parameters

Parameter	Type	Description
`account_name`		Azure storage account name.
`account_key`		Storage account access key (plain string or ``SecretStr``).
`container`		Blob container (equivalent to a bucket).

__init__

def __init__(
    account_name: str,
    account_key: str | SecretStr,
    container: str
) -> None

source

Initialise the Azure Blob Storage driver.

Parameters

Parameter	Type	Description
`account_name`	str	Azure storage account name.
`account_key`	str | SecretStr	Storage account access key.
`container`	str	Blob container name.

Raises

Exception	Description
ImportError	When ``azure-storage-blob`` is not installed.

upload

async def upload(
    path: str,
    data: Uploadable,
    content_type: str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to Azure Blob Storage at path.

Parameters

Parameter	Type	Description
`path`	str	Destination blob name / key.
`data`	Uploadable	Content to upload.
`content_type`	str | None	Optional MIME type override. **options: Additional upload options.

Returns

Type	Description
FileInfo	FileInfo for the stored blob.

Raises

Exception	Description
StorageError	On any Azure SDK error.

download

async def download(path: str) -> bytes

source

Download the blob at path into memory.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.

Returns

Type	Description
bytes	Raw file bytes.

Raises

Exception	Description
FileNotFoundError	When the blob does not exist.
StorageError	On any other Azure SDK error.

stream

async def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncGenerator[bytes, None]

source

Yield successive chunks from the Azure blob at path.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.
`chunk_size`	int	Bytes per chunk (default 8 KiB).

Yields

Type	Description
AsyncGenerator[bytes, None]	Raw byte chunks.

delete

async def delete(path: str) -> None

source

Delete the Azure blob at path.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.

Raises

Exception	Description
FileNotFoundError	When the blob does not exist.
StorageError	On any other Azure SDK error.

exists

async def exists(path: str) -> bool

source

Return True if path exists in the Azure container.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.

info

async def info(path: str) -> FileInfo

source

Return metadata for the Azure blob at path.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.

Returns

Type	Description
FileInfo	FileInfo.

Raises

Exception	Description
FileNotFoundError	When the blob does not exist.
StorageError	On any other Azure SDK error.

list

async def list(prefix: str = '') -> AsyncGenerator[FileInfo, None]

source

Yield FileInfo for blobs matching prefix.

Parameters

Parameter	Type	Description
`prefix`	str	Blob name prefix filter (empty string lists all blobs).

Yields

Type	Description
AsyncGenerator[FileInfo, None]	FileInfo entries.

Raises

Exception	Description
StorageError	On any Azure SDK error.

get_url

async def get_url(path: str) -> str

source

Return a public URL for the Azure blob at path.

The URL is only accessible when the container has public access enabled. For private containers, use get_presigned_url.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Return a SAS (Shared Access Signature) URL for the blob at path.

Parameters

Parameter	Type	Description
`path`	str	Blob name / key.
`expires_in`	timedelta	Validity window (default one hour).
`method`	str	HTTP verb (``"GET"`` for read, ``"PUT"`` for write).

Returns

Type	Description
str	A time-limited SAS URL string.

Raises

Exception	Description
StorageError	When SAS token generation fails.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check against Azure Blob Storage.

Lists at most one blob from the container to verify credentials and container accessibility.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

---

FileDeletedEvent

source

File was successfully deleted from storage.

Consumed by: file cleanup tracking, audit logging, quota reclamation.

---

FileDownloadedEvent

source

File was successfully downloaded from storage.

Consumed by: download tracking, audit logging, usage analytics.

---

FileInfo

source

Information about a stored file.

---

FileUploadedEvent

source

File was successfully uploaded to storage.

Consumed by: file tracking, audit logging, quota management.

---

GCSDriver

source

Google Cloud Storage driver.

Wraps gcloud-aio-storage to provide fully async GCS operations.

Install the optional dependency with

pip install lexigram-storage[gcs]
# i.e.  gcloud-aio-storage>=9.0.0

pip install lexigram-storage[gcs]
# i.e.  gcloud-aio-storage>=9.0.0

Parameters

Parameter	Type	Description
`bucket`		GCS bucket name.
`project_id`		Google Cloud project ID (used for logging / context only; the gcloud library resolves the project from ADC when needed).
`credentials_path`		Path to a service-account JSON key file. When ``None`` the driver falls back to Application Default Credentials (ADC), which covers Cloud Run, GKE Workload Identity, etc.

__init__

def __init__(
    bucket: str,
    project_id: str | None = None,
    credentials_path: str | None = None
) -> None

source

Initialise the GCS driver.

Parameters

Parameter	Type	Description
`bucket`	str	GCS bucket name.
`project_id`	str | None	Google Cloud project ID (informational; ADC handles auth).
`credentials_path`	str | None	Path to service-account JSON credentials file, or ``None`` to use Application Default Credentials.

Raises

Exception	Description
ImportError	When ``gcloud-aio-storage`` is not installed.

upload

async def upload(
    path: str,
    data: Uploadable,
    content_type: str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to the GCS path.

Parameters

Parameter	Type	Description
`path`	str	Destination object name / key.
`data`	Uploadable	Content to upload.
`content_type`	str | None	Optional MIME type override. **options: Additional upload options.

Returns

Type	Description
FileInfo	FileInfo for the stored object.

Raises

Exception	Description
StorageError	On any GCS API error.

download

async def download(path: str) -> bytes

source

Download the object at path into memory.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.

Returns

Type	Description
bytes	Raw file bytes.

Raises

Exception	Description
FileNotFoundError	When the object does not exist.
StorageError	On any other GCS API error.

stream

async def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncGenerator[bytes, None]

source

Yield successive chunks from the GCS object at path.

Downloads the entire object into memory first, then yields chunks from the in-memory buffer. For objects larger than available RAM, prefer using a streaming-capable GCS library.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.
`chunk_size`	int	Bytes per chunk (default 8 KiB).

Yields

Type	Description
AsyncGenerator[bytes, None]	Raw byte chunks.

delete

async def delete(path: str) -> None

source

Delete the GCS object at path.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.

Raises

Exception	Description
FileNotFoundError	When the object does not exist.
StorageError	On any other GCS API error.

exists

async def exists(path: str) -> bool

source

Return True if path exists in the GCS bucket.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.

info

async def info(path: str) -> FileInfo

source

Return metadata for the GCS object at path.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.

Returns

Type	Description
FileInfo	FileInfo.

Raises

Exception	Description
FileNotFoundError	When the object does not exist.
StorageError	On any other GCS API error.

list

async def list(prefix: str = '') -> AsyncGenerator[FileInfo, None]

source

Yield FileInfo for objects matching prefix.

Parameters

Parameter	Type	Description
`prefix`	str	Key prefix filter (empty string lists all objects).

Yields

Type	Description
AsyncGenerator[FileInfo, None]	FileInfo entries.

Raises

Exception	Description
StorageError	On any GCS API error.

get_url

async def get_url(path: str) -> str

source

Return a public URL for the GCS object at path.

The URL is only accessible when the object (or bucket) has been made publicly readable. For private objects, use get_presigned_url.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Return a signed URL for the GCS object at path.

Uses the V4 signing API from gcloud-aio-storage when available, falling back to an unsigned public URL for read access.

Parameters

Parameter	Type	Description
`path`	str	Object name / key.
`expires_in`	timedelta	Validity window (default one hour).
`method`	str	HTTP verb (``"GET"`` or ``"PUT"``).

Returns

Type	Description
str	A signed (or public) URL string.

Raises

Exception	Description
StorageError	When URL signing fails.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check against GCS.

Lists at most one object from the bucket to verify credentials and bucket accessibility.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

---

LocalDriver

source

Local file system storage driver

__init__

def __init__(
    root_dir: str = './storage',
    base_url: str = 'http://localhost:8000/storage'
)

source

upload

async def upload(
    path: str,
    data: Uploadable,
    content_type: str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to local file system with atomic write guarantee.

Uses temp file + atomic rename to ensure no partial writes on failure. Optionally validates checksum if provided in options.

download

async def download(path: str) -> bytes

source

Download file content from local file system

stream

async def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncGenerator[bytes, None]

source

Stream file content from local file system

delete

async def delete(path: str) -> None

source

Delete file from local file system

exists

async def exists(path: str) -> bool

source

Check if file exists in local file system

info

async def info(path: str) -> FileInfo

source

Get file info from local file system

list

async def list(prefix: str = '') -> AsyncGenerator[FileInfo, None]

source

List files with prefix from local file system

get_url

async def get_url(path: str) -> str

source

Get public URL for local file

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Get pre-signed URL for local file (same as public URL).

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check on local storage

copy

async def copy(
    src: str,
    dst: str
) -> FileInfo

source

Copy a file efficiently within the local filesystem.

move

async def move(
    src: str,
    dst: str
) -> FileInfo

source

Move a file efficiently within the local filesystem.

---

MemoryDriver

source

In-memory storage driver using a dictionary

__init__

def __init__() -> None

source

upload

async def upload(
    path: str,
    data: Uploadable,
    content_type: str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to memory storage

download

async def download(path: str) -> bytes

source

Download file content from memory

stream

async def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncGenerator[bytes, None]

source

Stream file content from memory

delete

async def delete(path: str) -> None

source

Delete file from memory

exists

async def exists(path: str) -> bool

source

Check if file exists in memory

info

async def info(path: str) -> FileInfo

source

Get file info from memory

list

async def list(prefix: str = '') -> AsyncGenerator[FileInfo, None]

source

List files with prefix from memory

get_url

async def get_url(path: str) -> str

source

Get URL for memory storage (not applicable)

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Raise StorageUnsupportedOperationError — not applicable to in-memory storage.

The in-memory driver has no network endpoint, so presigned URLs cannot be generated. Use a cloud-backed driver (S3, GCS, Azure) for presigned URL support.

Parameters

Parameter	Type	Description
`path`	str	File path that would have been signed.
`expires_in`	timedelta	Ignored.
`method`	str	Ignored.

Raises

Exception	Description
StorageUnsupportedOperationError	Always — presigned URLs are not supported by the in-memory driver.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check on memory storage

---

NamedStorageConfig

source

Configuration for a single named storage backend.

Used in StorageConfig.backends to declare multiple blob stores that the framework registers as named DI bindings.

Example

backends:

• name: primary driver: s3 primary: true s3: bucket: my-app-primary region: us-east-1
• name: avatars driver: s3 s3: bucket: my-app-avatars region: us-east-1
• name: local driver: local local: root_dir: ./storage/public

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed BlobStoreProtocol binding.
`driver`		Storage driver name. One of: local, s3, gcs, azure, r2, memory.
`local`		Local filesystem driver config (when driver='local').
`s3`		AWS S3 driver config (when driver='s3').
`gcs`		Google Cloud Storage config (when driver='gcs').
`azure`		Azure Blob Storage config (when driver='azure').
`r2`		Cloudflare R2 config (when driver='r2').
`memory`		In-memory storage config (when driver='memory').

---

ObjectDeletedHook

source

Payload fired when a blob object is deleted.

Attributes: bucket: Name of the bucket or container the object was deleted from. key: Storage key (path) of the deleted object.

---

ObjectStoredHook

source

Payload fired when a blob object is successfully stored.

Attributes: bucket: Name of the bucket or container the object was stored in. key: Storage key (path) of the stored object.

---

S3Driver

source

AWS S3 storage driver with multipart upload for large files

__init__

def __init__(
    bucket: str,
    region: str | None = None,
    access_key: str | None = None,
    secret_key: str | None = None,
    endpoint_url: str | None = None,
    public_url: str | None = None,
    multipart_threshold: int = DEFAULT_MULTIPART_THRESHOLD,
    multipart_chunk_size: int = DEFAULT_MULTIPART_CHUNK_SIZE,
    encryption: EncryptionConfig | None = None
)

source

Initialize S3 driver.

Parameters

Parameter	Type	Description
`bucket`	str	S3 bucket name
`region`	str | None	AWS region
`access_key`	str | None	AWS access key (optional if using IAM roles)
`secret_key`	str | None	AWS secret key (optional if using IAM roles)
`endpoint_url`	str | None	Custom endpoint URL (for MinIO, LocalStack, R2 API, etc.)
`public_url`	str | None	Custom public URL for serving files (e.g., R2 custom domain). When set, get_url() returns this instead of the endpoint_url.
`multipart_threshold`	int	File size threshold for multipart upload (bytes)
`multipart_chunk_size`	int	Chunk size for multipart upload (bytes)
`encryption`	EncryptionConfig | None	Optional server-side encryption configuration. When ``enabled=True``, all ``put_object`` and ``create_multipart_upload`` calls will include the appropriate ``ServerSideEncryption`` / ``SSEKMSKeyId`` parameters.

download

async def download(path: str) -> bytes

source

Download file content from S3 (robust to different body shapes).

Some S3 client implementations return an async context manager for the response body while others provide an object with a read() coroutine method. Handle both shapes defensively and return bytes.

stream

async def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncGenerator[bytes, None]

source

Stream file content from S3

delete

async def delete(path: str) -> None

source

Delete file from S3

exists

async def exists(path: str) -> bool

source

Check if file exists in S3

info

async def info(path: str) -> FileInfo

source

Get file info from S3

list

async def list(prefix: str = '') -> AsyncGenerator[FileInfo, None]

source

List files with prefix from S3

get_url

async def get_url(path: str) -> str

source

Get public URL for S3 object

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Get pre-signed URL for S3 object.

Parameters

Parameter	Type	Description
`path`	str	Storage path / object key.
`expires_in`	timedelta	Validity window (default one hour).
`method`	str	HTTP verb — ``"GET"`` or ``"PUT"``.

Returns

Type	Description
str	Pre-signed URL string.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check on S3 storage.

---

StorageConfig

source

Hierarchical root configuration for Lexigram Storage.

Attributes: name: Configuration name (default: “storage”) enabled: Whether storage module is enabled default_driver: Default storage driver (local, s3, gcs, azure, memory) drivers: Driver-specific configurations service: Storage operation settings

Example

# From environment variables (``LEX_STORAGE__*``)
config = StorageConfig()

# Explicit values
config = StorageConfig(default_driver="s3", drivers={"s3": StorageS3Config(...)})

# From environment variables (``LEX_STORAGE__*``)
config = StorageConfig()

# Explicit values
config = StorageConfig(default_driver="s3", drivers={"s3": StorageS3Config(...)})

is_production

property is_production() -> bool

source

Check if running in production environment.

is_development

property is_development() -> bool

source

Check if running in development environment.

is_test

property is_test() -> bool

source

Check if running in test environment.

from_named

def from_named(
    cls,
    entry: NamedStorageConfig
) -> StorageConfig

source

Build a single-backend StorageConfig from a NamedStorageConfig entry.

Used internally by StorageProvider to create per-backend configs from a multi-backend declaration. The resulting config has the driver-specific config placed in the drivers dict under the driver type key, so the existing DriverRegistry can consume it unchanged.

Parameters

Parameter	Type	Description
`entry`	NamedStorageConfig	The named backend entry to materialise.

Returns

Type	Description
StorageConfig	A StorageConfig configured for the single named backend.

validate_production_security

def validate_production_security() -> StorageConfig

source

Block insecure storage configurations in production.

This validator fires when the LEX_ENV environment variable is set to "production" (case-insensitive). It rejects known-weak placeholder credentials ("change-me", "password", etc.) in S3 and Azure driver configs, raising ValueError immediately so the application fails fast at startup rather than leaking credentials at request time.

The environment variable checked is LEX_ENV (default: "development"). Set LEX_ENV=production in your deployment environment to activate production-grade security checks.

---

StorageModule

source

Object storage backend (S3, GCS, Azure, R2, local filesystem).

Registers BlobStoreProtocol for constructor injection.

Call configure to configure and register a storage driver, or stub for an isolated in-memory setup with no external service dependencies.

Usage

from lexigram.storage.config import StorageConfig

@module(
    imports=[StorageModule.configure(StorageConfig(default_driver="s3"))]
)
class AppModule(Module):
    pass

from lexigram.storage.config import StorageConfig

@module(
    imports=[StorageModule.configure(StorageConfig(default_driver="s3"))]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: StorageConfig | Any | None = None,
    enable_encryption: bool = False
) -> DynamicModule

source

Create a StorageModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	StorageConfig | Any | None	StorageConfig or ``None`` to resolve config from the container at boot time.
`enable_encryption`	bool	Enable server-side encryption for all stored objects. Defaults to ``False``; configure encryption keys via EncryptionConfig.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

Raises

Exception	Description
TypeError	If *config* is not a ``StorageConfig`` or ``None``.

stub

def stub(
    cls,
    config: StorageConfig | None = None
) -> DynamicModule

source

Create a StorageModule suitable for unit and integration testing.

Uses an in-memory backend with no external service dependencies.

Parameters

Parameter	Type	Description
`config`	StorageConfig | None	Optional StorageConfig override. Uses safe in-memory defaults when ``None``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

StorageProvider

source

DI provider for storage services.

Registers both the DriverRegistry (for extensibility) and the configured BlobStoreProtocol driver singleton into the container.

Usage

# Pass config directly
app.add_provider(StorageProvider(config=StorageConfig()))

# Or let the provider resolve config from the container (requires ConfigProvider)
app.add_provider(StorageProvider())

# Later in a service:
class MyService:
    def __init__(self, store: BlobStoreProtocol) -> None: ...

# Pass config directly
app.add_provider(StorageProvider(config=StorageConfig()))

# Or let the provider resolve config from the container (requires ConfigProvider)
app.add_provider(StorageProvider())

# Later in a service:
class MyService:
    def __init__(self, store: BlobStoreProtocol) -> None: ...

__init__

def __init__(config: StorageConfig | None = None) -> None

source

config

property config() -> StorageConfig | None

source

Return the storage configuration.

config

def config(value: StorageConfig | None) -> None

source

Set the storage configuration.

from_config

def from_config(
    cls,
    config: StorageConfig,
    **context: Any
) -> Self

source

Create provider from config object.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register storage services into the DI container.

If no config was supplied at construction time, resolves it from the container via ConfigProtocol. Binds:

• DriverRegistry — the extensible driver factory (singleton).
• BlobStoreProtocol — the configured backend driver instance (singleton). In multi-backend mode, also registers named bindings.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	DI registrar supplied by the framework.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Startup hook — verify storage connectivity after all registrations.

shutdown

async def shutdown() -> None

source

Shutdown hook - cleanup resources owned by driver.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Health check for storage.

---

UploadOptions

source

Options for file uploads.

---