S3-compatible object storage

Off-load large task outputs to S3, MinIO, R2, or any S3-compatible store.

Flux 0.56.0 ships two output-storage backends:

There is no S3 backend in core. The only mention of "s3" in flux/output_storage.py is a docstring example for OutputStorageReference. Adding one is straightforward — OutputStorage is a three-method abstract base class — and that recipe is below.

The contract

From flux/output_storage.py:

class OutputStorage(ABC):
    @abstractmethod
    def store(self, reference_id: str, value: Any) -> OutputStorageReference: ...
    @abstractmethod
    def retrieve(self, reference: OutputStorageReference) -> Any: ...
    @abstractmethod
    def delete(self, reference: OutputStorageReference) -> Any: ...

OutputStorageReference has three fields: storage_type (a string discriminator), reference_id (whatever your backend needs to find the object later), and metadata (a dict — LocalFileStorage uses it to record which serializer wrote the bytes).

Recipe: an S3 backend

from __future__ import annotations
import dill
import json
from typing import Any

import boto3
from botocore.exceptions import ClientError

from flux.output_storage import OutputStorage, OutputStorageReference


class S3OutputStorage(OutputStorage):
    storage_type = "s3"

    def __init__(self, bucket: str, prefix: str = "flux/outputs",
                 client: Any | None = None, serializer: str = "pkl"):
        self.bucket = bucket
        self.prefix = prefix.strip("/")
        self.client = client or boto3.client("s3")
        self.serializer = serializer

    def _key(self, reference_id: str) -> str:
        return f"{self.prefix}/{reference_id}.{self.serializer}"

    def _serialize(self, value: Any) -> bytes:
        if self.serializer == "json":
            return json.dumps(value).encode("utf-8")
        return dill.dumps(value)

    def _deserialize(self, blob: bytes, serializer: str | None = None) -> Any:
        s = serializer or self.serializer
        return json.loads(blob) if s == "json" else dill.loads(blob)

    def store(self, reference_id: str, value: Any) -> OutputStorageReference:
        body = self._serialize(value)
        key = self._key(reference_id)
        self.client.put_object(Bucket=self.bucket, Key=key, Body=body)
        return OutputStorageReference(
            storage_type=self.storage_type,
            reference_id=reference_id,
            metadata={"bucket": self.bucket, "key": key, "serializer": self.serializer},
        )

    def retrieve(self, reference: OutputStorageReference) -> Any:
        if reference.storage_type != self.storage_type:
            raise ValueError(f"Invalid storage type: {reference.storage_type}")
        meta = reference.metadata
        obj = self.client.get_object(Bucket=meta["bucket"], Key=meta["key"])
        return self._deserialize(obj["Body"].read(), meta.get("serializer"))

    def delete(self, reference: OutputStorageReference) -> Any:
        if reference.storage_type != self.storage_type:
            raise ValueError(f"Invalid storage type: {reference.storage_type}")
        meta = reference.metadata
        try:
            self.client.delete_object(Bucket=meta["bucket"], Key=meta["key"])
        except ClientError:
            pass

Attach it to a task with output_storage:

from flux import task

storage = S3OutputStorage(bucket="my-flux-outputs")

@task.with_options(output_storage=storage)
async def extract_dataframe(...) -> pd.DataFrame:
    ...

storage needs to be importable on every worker — define it in a module shared with your workflows, not inside if __name__ == "__main__".

S3-compatible providers

The same class points at any S3-API store; only the boto3.client(...) configuration changes:

For everything except AWS S3, store the credentials in Flux’s secret store and instantiate the client from inside the storage class rather than at module import — secrets are not resolved until a task runs.

What goes wrong

See also

Derived against Flux 0.56.0 and boto3 1.35.x, 2026-07.