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: Uploadable,
    content_type: UploadOptions | str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to the storage backend.

Parameters

Parameter	Type	Description
path	str	Storage path/key.
data	Uploadable	File content as bytes, text, file-like object, or async iterator.
content_type	UploadOptions | str | None	MIME type or UploadOptions for metadata and other upload settings. **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 quadkit-storage[azure]
# i.e.  azure-storage-blob>=12.20.0

pip install quadkit-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: UploadOptions | 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	UploadOptions | 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.

Returns

Type	Description
bool	`False` when the blob does not exist.

Raises

Exception	Description
StorageError	On any other Azure SDK error — a failed connectivity/credentials check must not be reported as "does not exist" (the old narrow except tuple also missed Azure SDK error classes, which are not OSError subclasses).

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 quadkit-storage[gcs]
# i.e.  gcloud-aio-storage>=9.0.0

pip install quadkit-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: UploadOptions | 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	UploadOptions | 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.

Returns

Type	Description
bool	`False` when the object does not exist.

Raises

Exception	Description
StorageError	On any other GCS API error — a failed connectivity/credentials check must not be reported as "does not exist" (the old narrow except tuple also missed google-api-core error classes, which are not OSError subclasses).

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: UploadOptionsContract | 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 a pre-signed URL for local file.

Local driver does not support presigned URLs; callers must fall back to the public URL via get_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: UploadOptionsContract | 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 Quadkit 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 (``QK_STORAGE__*``)
config = StorageConfig()

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

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

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

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 environment is set to production. 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.

---

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 quadkit.storage.config import StorageConfig

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

from quadkit.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.

---