Storage

Expanse provides a simple and unified API for managing files and directories across different storage backends. It supports local filesystem storage, as well as cloud storage services like Amazon S3.

The idea is that you can switch between storage services without changing your application code. During development, you might want to store your files in your local filesystem but in production, you will most likely want switch to a cloud provider. All of this can be done by only changing a few environment variables, without having to change a single line of code in your application.

Configuration

The storage system is configured based on two concepts: drivers and storages. Drivers are the underlying implementations of the storage system, like local filesystem or Amazon S3, while storages represent a storage layer supported by one of the supported drivers. You can have multiple storages using the same driver, and each storage can be configured independently.

To start using the storage system, you need to configure at least one storage and its corresponding driver and reference it in the STORAGE_STORAGE environment variable.

STORAGE_STORAGE=default

Each storage is configured via environment variables following the pattern STORAGE_STORAGES__STORAGE_NAME__PARAMETER=VALUE, where STORAGE_NAME is the name of the storage and PARAMETER is a parameter supported by the driver used by the storage.

Available drivers

The following drivers are currently supported:

Amazon S3

To use Amazon S3 as a storage backend, you need to configure a storage with the s3 driver and provide the necessary parameters for connecting to your S3 bucket.

STORAGE_STORAGE=s3
STORAGE_STORAGES__S3__DRIVER=s3
STORAGE_STORAGES__S3__KEY=your-access-key-id
STORAGE_STORAGES__S3__SECRET=your-secret-access-key
STORAGE_STORAGES__S3__REGION=your-region
STORAGE_STORAGES__S3__BUCKET=your-bucket-name
STORAGE_STORAGES__S3__ENDPOINT=

Local filesystem

To use the local filesystem as a storage backend, you need to configure a storage with the local driver and provide the necessary parameters for connecting to your local filesystem.

STORAGE_STORAGE=local
STORAGE_STORAGES__LOCAL__DRIVER=local
STORAGE_STORAGES__LOCAL__ROOT=storage/app/files

Using the storage system

To use the storage system in your application, you can inject a Storage instance anywhere dependency injection is supported, such as in controllers, middleware, jobs, message handlers and beam commands.

from expanse.contracts.asynchronous.storage import Storage


async def upload_file(storage: Storage) -> None:
    ...

from expanse.contracts.synchronous.storage import Storage


def upload_file(storage: Storage) -> None:
    ...

If you have configured multiple storages, you can specify which storage to use by injecting a Storage instance annotated with the storage name.

from typing import Annotated
from expanse.contracts.asynchronous.storage import Storage


async def upload_file(storage: Annotated[Storage, "s3"]) -> None:
    ...

from typing import Annotated
from expanse.contracts.synchronous.storage import Storage


def upload_file(storage: Annotated[Storage, "s3"]) -> None:
    ...

Retrieving files

To retrieve a file from the storage, you can use the get method of the Storage instance, which returns the content of the file as bytes.

from expanse.contracts.asynchronous.storage import Storage


async def download_file(storage: Storage) -> None:
    content = await storage.get("path/to/file.txt")
    ...

from expanse.contracts.synchronous.storage import Storage


def download_file(storage: Storage) -> None:
    content = storage.get("path/to/file.txt")
    ...

Since the get method returns the content of the file as bytes, it is suitable for small files. For larger files, you can use the stream method, which returns an asynchronous or synchronous generator that yields the content of the file in chunks.

from expanse.contracts.asynchronous.storage import Storage


async def download_file(storage: Storage) -> None:
    async for chunk in (await storage.stream("path/to/file.txt")):
        ...

from expanse.contracts.synchronous.storage import Storage


def download_file(storage: Storage) -> None:
    for chunk in storage.stream("path/to/file.txt"):
        ...

Storing files

To store a file in the storage, you can use the put method of the Storage instance, which accepts the a path and the file content as bytes or a file-like object.

from expanse.contracts.asynchronous.storage import Storage


async def upload_file(storage: Storage) -> None:
    content = b"file content"
    await storage.put("path/to/file.txt", content)

from expanse.contracts.synchronous.storage import Storage


def upload_file(storage: Storage) -> None:
    content = b"file content"
    storage.put("path/to/file.txt", content)

If you have a file-like object, you can pass it directly to the put method without reading its content into memory:

from expanse.contracts.asynchronous.storage import Storage


async def upload_file(storage: Storage) -> None:
    with open("path/to/local/file.txt", "rb") as file:
        await storage.put("path/to/file.txt", file)

from expanse.contracts.synchronous.storage import Storage


def upload_file(storage: Storage) -> None:
    with open("path/to/local/file.txt", "rb") as file:
        storage.put("path/to/file.txt", file)

Uploading files

If you need to storage a file uploaded by the user, you can use the save method of an UploadFile instance to store it in the configured storage.

from expanse.contracts.asynchronous.storage import Storage
from expanse.http.upload_file import UploadFile


async def upload_file(storage: Storage, avatar: UploadFile) -> None:
    await my_file.save("avatars")

from expanse.contracts.synchronous.storage import Storage
from expanse.http.upload_file import UploadFile


def upload_file(storage: Storage, avatar: UploadFile) -> None:
    my_file.save("avatars")

save() takes the directory where the file should be stored as a parameter and returns the path of the stored file. The filename will be automatically generated to ensure uniqueness, but you can also specify a custom filename by using the name keyword argument.

from expanse.contracts.asynchronous.storage import Storage
from expanse.http.upload_file import UploadFile


async def upload_file(storage: Storage, avatar: UploadFile) -> None:
    await my_file.save("avatars", name="custom_filename.txt")

from expanse.contracts.synchronous.storage import Storage
from expanse.http.upload_file import UploadFile


def upload_file(storage: Storage, avatar: UploadFile) -> None:
    my_file.save("avatars", name="custom_filename.txt")

If you'd like to store the file in a specific storage, you can specify the name of the storage via the storage keyword argument:

from expanse.contracts.asynchronous.storage import Storage
from expanse.http.upload_file import UploadFile


async def upload_file(avatar: UploadFile) -> None:
    await my_file.save("avatars", storage="s3")

from expanse.contracts.synchronous.storage import Storage
from expanse.http.upload_file import UploadFile


def upload_file(avatar: UploadFile) -> None:
    my_file.save("avatars", storage="s3")

Deleting files

To delete a file from the storage, you can use the delete method of the Storage instance, which accepts the path of the file to be deleted.

from expanse.contracts.asynchronous.storage import Storage


async def delete_file(storage: Storage) -> None:
    await storage.delete("path/to/file.txt")

from expanse.contracts.synchronous.storage import Storage


def delete_file(storage: Storage) -> None:
    storage.delete("path/to/file.txt")