Storage Plugin

The StoragePlugin provides seamless integration with Litestar applications, handling dependency injection and lifecycle management for storage instances.

Plugin Class

class litestar_storages.contrib.plugin.StoragePlugin[source]

Bases: InitPluginProtocol

Litestar plugin for storage integration.

This plugin provides dependency injection support for storage instances, enabling seamless integration with Litestar applications. It supports both single and multiple named storage configurations.

Provides:

  • Dependency injection of storage instances

  • Lifespan management (connection cleanup)

  • Multiple named storage support

Example

Single storage configuration:

from litestar import Litestar
from litestar_storages import S3Storage, S3Config, StoragePlugin

storage = S3Storage(config=S3Config(bucket="uploads"))

app = Litestar(
    route_handlers=[...],
    plugins=[StoragePlugin(storage)],
)

Multiple named storages:

from litestar import Litestar
from litestar_storages import S3Storage, AzureStorage, StoragePlugin

app = Litestar(
    route_handlers=[...],
    plugins=[
        StoragePlugin(
            default=S3Storage(config=S3Config(bucket="main-uploads")),
            images=S3Storage(config=S3Config(bucket="images")),
            documents=AzureStorage(config=AzureConfig(container="docs")),
        )
    ],
)

Using in route handlers:

from litestar import post
from litestar.datastructures import UploadFile
from litestar_storages import Storage, StoredFile


@post("/upload")
async def upload(
    data: UploadFile,
    storage: Storage,  # Injected default storage
) -> StoredFile:
    return await storage.put(
        key=f"uploads/{data.filename}",
        data=data.file,
        content_type=data.content_type,
    )


@post("/upload-image")
async def upload_image(
    data: UploadFile,
    images_storage: Storage,  # Injected named storage
) -> StoredFile:
    return await images_storage.put(
        key=f"images/{data.filename}",
        data=data.file,
        content_type=data.content_type,
    )
Parameters:
__init__(default=None, **named_storages)[source]

Initialize the StoragePlugin.

Parameters:

  • default (Storage | None) – Optional default storage instance. If provided, it will be registered as the “storage” dependency.

  • **named_storages (Storage) – Named storage instances. Each will be registered as “{name}_storage” dependency.

Example:

plugin = StoragePlugin(
    default=S3Storage(...),
    images=S3Storage(...),
    documents=AzureStorage(...),
)
This registers three dependencies:

  • storage (from default)

  • images_storage

  • documents_storage

storages: dict[str, Storage]
on_app_init(app_config)[source]

Register storage instances as dependencies on application initialization.

This method is called by Litestar during application startup. It registers each storage instance as a dependency that can be injected into route handlers.

Parameters:

app_config (AppConfig) – The Litestar application configuration

Return type:

AppConfig

Returns:

Modified application configuration with storage dependencies registered

Note

  • The “default” storage is registered as “storage”

  • Named storages are registered as “{name}_storage”

  • Existing dependencies are preserved

Features

Dependency Injection

The plugin registers storage instances as dependencies that can be injected into route handlers:

from litestar import Litestar, get, post
from litestar_storages import Storage, StoredFile
from litestar_storages.contrib import StoragePlugin

@post("/upload")
async def upload(
    data: bytes,
    storage: Storage,  # Injected
) -> StoredFile:
    return await storage.put("file.txt", data)

@get("/files")
async def list_files(storage: Storage) -> list[StoredFile]:
    return [file async for file in storage.list()]

Named Storages

Register multiple storages with different names:

plugin = StoragePlugin(
    default=s3_storage,      # Injected as "storage"
    uploads=upload_storage,  # Injected as "uploads_storage"
    cache=cache_storage,     # Injected as "cache_storage"
)

@post("/upload")
async def upload(
    storage: Storage,         # Default storage
    uploads_storage: Storage, # Named storage
) -> StoredFile:
    ...

Automatic Cleanup

The plugin registers a shutdown handler that closes all storage instances:

# On application shutdown, this is called automatically:
for storage in storages.values():
    await storage.close()

This ensures proper cleanup of:

  • HTTP sessions (aioboto3, gcloud-aio-storage, azure-storage-blob)

  • Connection pools

  • File handles

Usage Examples

Single Storage

from litestar import Litestar
from litestar_storages import S3Storage, S3Config
from litestar_storages.contrib import StoragePlugin

storage = S3Storage(
    config=S3Config(
        bucket="my-uploads",
        region="us-east-1",
    )
)

app = Litestar(
    route_handlers=[...],
    plugins=[StoragePlugin(default=storage)],
)

Multiple Storages by Environment

import os
from litestar import Litestar
from litestar_storages import (
    MemoryStorage,
    S3Storage, S3Config,
    FileSystemStorage, FileSystemConfig,
)
from litestar_storages.contrib import StoragePlugin

def get_storage():
    env = os.getenv("ENV", "development")

    if env == "production":
        return S3Storage(config=S3Config(
            bucket=os.environ["S3_BUCKET"],
            region=os.environ["AWS_REGION"],
        ))
    elif env == "staging":
        return FileSystemStorage(config=FileSystemConfig(
            path=Path("/var/uploads"),
        ))
    else:
        return MemoryStorage()

app = Litestar(
    route_handlers=[...],
    plugins=[StoragePlugin(default=get_storage())],
)

With DTOs

from litestar import post
from litestar.datastructures import UploadFile
from litestar_storages import Storage, StoredFile
from litestar_storages.contrib import StoragePlugin, StoredFileDTO

@post("/upload", return_dto=StoredFileDTO)
async def upload(
    data: UploadFile,
    storage: Storage,
) -> StoredFile:
    content = await data.read()
    return await storage.put(
        key=f"uploads/{data.filename}",
        data=content,
        content_type=data.content_type,
    )

Complete Application

from litestar import Litestar, get, post, delete
from litestar.datastructures import UploadFile
from litestar.response import Stream
from litestar_storages import Storage, StoredFile
from litestar_storages.contrib import StoragePlugin, StoredFileDTO
from litestar_storages.exceptions import StorageFileNotFoundError

@post("/files", return_dto=StoredFileDTO)
async def upload_file(
    data: UploadFile,
    storage: Storage,
) -> StoredFile:
    content = await data.read()
    return await storage.put(
        key=f"files/{data.filename}",
        data=content,
        content_type=data.content_type,
    )

@get("/files/{key:path}")
async def download_file(
    key: str,
    storage: Storage,
) -> Stream:
    async def stream():
        async for chunk in storage.get(f"files/{key}"):
            yield chunk
    return Stream(stream())

@get("/files/{key:path}/info", return_dto=StoredFileDTO)
async def get_file_info(
    key: str,
    storage: Storage,
) -> StoredFile:
    return await storage.info(f"files/{key}")

@delete("/files/{key:path}")
async def delete_file(
    key: str,
    storage: Storage,
) -> None:
    await storage.delete(f"files/{key}")

app = Litestar(
    route_handlers=[upload_file, download_file, get_file_info, delete_file],
    plugins=[StoragePlugin(default=S3Storage(config=S3Config(bucket="files")))],
)