Configuring AWS S3

When working with remote datasets or artifacts in AWS S3 or S3-compatible storage services (such as MinIO, Ceph, or LocalStack), flyte-sdk provides the flyte.storage.S3 configuration class to manage endpoints, authentication credentials, retry policies, and low-level filesystem options.

Under the hood, storage interactions rely on obstore and fsspec. The flyte.storage.S3 class encapsulates S3-specific connection settings and converts them into the appropriate parameters expected by the filesystem backend.

Configuring S3 Programmatically

To configure AWS S3 access explicitly in Python code, instantiate flyte.storage.S3 and pass it to flyte.init().

from datetime import timedelta
import flyte
from flyte.storage import S3

# Initialize S3 storage configuration
s3_storage = S3(
    access_key_id="YOUR_ACCESS_KEY_ID",
    secret_access_key="YOUR_SECRET_ACCESS_KEY",
    retries=5,
    backoff=timedelta(seconds=2),
    enable_debug=False,
)

# Register the storage backend with flyte-sdk
flyte.init(storage=s3_storage)

Configuration Parameters

The S3 dataclass accepts both provider-specific and general storage parameters:

Parameter	Type	Default	Description
access_key_id	str | None	None	AWS access key ID.
secret_access_key	str | None	None	AWS secret access key.
endpoint	str | None	None	Custom S3 endpoint URL (e.g., MinIO or LocalStack).
retries	int	3	Maximum number of retry attempts for failed storage calls.
backoff	timedelta	timedelta(seconds=5)	Initial backoff duration between retries.
enable_debug	bool	False	Toggles debug logging for storage operations.
attach_execution_metadata	bool	True	Whether execution metadata is attached to stored artifacts.

Automatic Configuration via Environment Variables

Instead of hardcoding credentials, you can configure S3 storage using environment variables. Call S3.auto() to construct an S3 instance from the runtime environment.

import flyte
from flyte.storage import S3

# Reads FLYTE_AWS_* and UNION_STORAGE_* environment variables
s3_storage = S3.auto()

flyte.init(storage=s3_storage)

Environment Variable Mapping

flyte.storage.S3 inspects the following environment variables:

• FLYTE_AWS_ACCESS_KEY_ID: S3 access key ID.
• FLYTE_AWS_SECRET_ACCESS_KEY: S3 secret access key.
• FLYTE_AWS_ENDPOINT: Endpoint URL for S3 or S3-compatible services.
• UNION_STORAGE_RETRIES: Maximum number of retry attempts (integer).
• UNION_STORAGE_BACKOFF_SECONDS: Initial retry backoff in seconds (integer/float).
• UNION_STORAGE_DEBUG: Enables storage debug logs (true or false).

export FLYTE_AWS_ACCESS_KEY_ID="AKIAIOSFODNN7EXAMPLE"
export FLYTE_AWS_SECRET_ACCESS_KEY="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
export UNION_STORAGE_RETRIES=5
export UNION_STORAGE_BACKOFF_SECONDS=3

If flyte.init() is called without a storage instance or if an uninitialized storage operation targets an s3:// path, flyte-sdk automatically falls back to S3.auto().

Connecting to Custom S3 Endpoints (MinIO, LocalStack)

When using self-hosted MinIO or custom S3-compatible object stores, specify the endpoint argument. The configuration automatically sets allow_http: True in the underlying client options to permit plain HTTP endpoints for local or on-premise development.

import flyte
from flyte.storage import S3

minio_storage = S3(
    endpoint="http://minio.local:9000",
    access_key_id="minioadmin",
    secret_access_key="minioadmin",
)

flyte.init(storage=minio_storage)

Using the Pre-Configured Sandbox Preset

For local development against the Flyte sandbox environment, S3.for_sandbox() returns a pre-configured S3 instance targeting http://localhost:4566 with standard local credentials (minio / miniostorage).

import flyte
from flyte.storage import S3

# Load preset configured for LocalStack/MinIO on localhost:4566
sandbox_storage = S3.for_sandbox()

flyte.init(storage=sandbox_storage)

Reading and Writing S3 Objects in Tasks

Once initialized, flyte.io.File, flyte.io.Dir, and direct storage primitives (flyte.storage.get, flyte.storage.put) automatically use your configured S3 settings for all s3:// URIs.

import flyte
from flyte.io import File
from flyte.storage import S3

flyte.init(storage=S3.auto())

@flyte.task
def process_s3_data(input_file: File) -> File:
    # Read file directly from S3
    with input_file.open("r") as f:
        content = f.read()
    
    transformed_content = content.upper()
    
    # Return local artifact or write back to S3
    output_path = "/tmp/output.txt"
    with open(output_path, "w") as f:
        f.write(transformed_content)
    
    return File(path=output_path)

You can also use low-level storage streaming functions:

import asyncio
from flyte import storage

async def stream_data():
    # Stream bytes directly to an S3 path
    data = b"Streaming S3 payload data"
    await storage.put_stream(data, to_path="s3://my-bucket/streamed.txt")

    # Read back as an async stream
    async for chunk in storage.get_stream("s3://my-bucket/streamed.txt", chunk_size=1024):
        print(len(chunk))

asyncio.run(stream_data())

Troubleshooting

Standard AWS Environment Variables Are Ignored

If credentials set in standard AWS_ACCESS_KEY_ID or AWS_SECRET_ACCESS_KEY are not recognized by S3.auto(), verify that you use the FLYTE_AWS_ prefix:

• Set FLYTE_AWS_ACCESS_KEY_ID instead of AWS_ACCESS_KEY_ID.
• Set FLYTE_AWS_SECRET_ACCESS_KEY instead of AWS_SECRET_ACCESS_KEY.
• Set FLYTE_AWS_ENDPOINT instead of AWS_ENDPOINT_URL.

Anonymous Access Fallbacks

When authenticating against public S3 buckets, flyte-sdk generates filesystem arguments with skip_signature=True under the hood if anonymous=True is passed to get_fsspec_kwargs(). If an authenticated read fails with a generic error or OS error on an unauthenticated bucket, flyte-sdk's internal storage layer will automatically attempt an anonymous retry.