Automatic Configuration via Environment Variables

Hardcoding storage credentials, custom bucket endpoints, and retry policies into task definitions creates security risks and makes deploying workflows across development, staging, and production environments error-prone. In flyte-sdk, storage configuration can be constructed dynamically from the environment using the .auto() class method available on the base Storage class and its cloud-specific implementations (S3, GCS, ABFS).

Configuring Storage from Environment Variables

To instantiate storage configuration without hardcoding credentials or connection details, call .auto() on the appropriate storage class defined in flyte.storage:

from flyte.storage import ABFS, GCS, S3, Storage

# Generic base configuration from UNION_STORAGE_* environment variables
base_storage = Storage.auto()

# AWS S3 / S3-compatible configuration
s3_storage = S3.auto()

# Google Cloud Storage configuration
gcs_storage = GCS.auto()

# Azure Blob Storage / ADLS Gen2 configuration
abfs_storage = ABFS.auto()

You can pass the instantiated object directly to flyte.init(storage=...) during system initialization:

import flyte
from flyte.storage import S3

# Initialize Flyte with automatically configured S3 storage
await flyte.init(
    endpoint="dns:///flyte.example.com",
    storage=S3.auto(),
)

How .auto() Works Internally

Each storage configuration class in storage/_config.py is an immutable, frozen @dataclass maintaining a class-level variable _KEY_ENV_VAR_MAPPING that maps dataclass fields to their corresponding environment variable names.

When .auto() is invoked:

1. The class queries environment variables using os.getenv.
2. Values are filtered through set_if_exists (from flyte.config), which only populates a key if the environment variable value is a boolean or a non-empty string.
3. The resulting keyword arguments are unpacked to construct an immutable instance of the dataclass. If an environment variable is not set or is empty, the dataclass retains its default value.

+-------------------------------------------------------------------------------+
|                             Environment Variables                             |
|  UNION_STORAGE_*  |  FLYTE_AWS_*  |  GCP_GSUTIL_*  |  AZURE_STORAGE_* / AZURE_*|
+-------------------------------------------------------------------------------+
                                      |
                                      v
                             os.getenv(...) lookup
                                      |
                                      v
                        set_if_exists(kwargs, key, val)
             (Filters out None and empty strings, preserves booleans)
                                      |
                                      v
                  Construct Frozen Dataclass (Storage / S3 / GCS / ABFS)
                                      |
                                      v
                          get_fsspec_kwargs(...)
                    (Emits parameters for fsspec / obstore)

Common Storage Settings (Storage)

Storage defines base configurations common across all storage backends:

@dataclass(init=True, repr=True, eq=True, frozen=True)
class Storage(object):
    retries: int = 3
    backoff: datetime.timedelta = datetime.timedelta(seconds=5)
    enable_debug: bool = False
    attach_execution_metadata: bool = True

Storage._auto_as_kwargs() parses the following environment variables:

Environment Variable	Dataclass Field	Default Value	Description
UNION_STORAGE_DEBUG	enable_debug	False	Enables verbose debug logging for storage operations.
UNION_STORAGE_RETRIES	retries	3	Maximum number of retry attempts for failed storage calls.
UNION_STORAGE_BACKOFF_SECONDS	backoff	timedelta(seconds=5)	Initial backoff delay between storage retry attempts.

Calling Storage.auto() returns a Storage instance populated with these environment variable overrides.

Provider-Specific Configurations

AWS S3 and S3-Compatible Backends (S3)

S3 inherits from Storage and adds support for custom endpoints (such as MinIO, Ceph, or LocalStack) and AWS credentials.

@dataclass(init=True, repr=True, eq=True, frozen=True)
class S3(Storage):
    endpoint: typing.Optional[str] = None
    access_key_id: typing.Optional[str] = None
    secret_access_key: typing.Optional[str] = None

S3.auto() combines the base Storage environment variables with AWS-specific settings:

Environment Variable	Dataclass Field	Default Value	Description
FLYTE_AWS_ENDPOINT	endpoint	None	Custom S3 endpoint URL (maps to endpoint_url in filesystem kwargs).
FLYTE_AWS_ACCESS_KEY_ID	access_key_id	None	AWS access key ID.
FLYTE_AWS_SECRET_ACCESS_KEY	secret_access_key	None	AWS secret access key.
UNION_STORAGE_DEBUG	enable_debug	False	Inherited from Storage.
UNION_STORAGE_RETRIES	retries	3	Inherited from Storage.
UNION_STORAGE_BACKOFF_SECONDS	backoff	timedelta(seconds=5)	Inherited from Storage.

Local Development with S3.for_sandbox()

For local sandbox or MinIO emulation environments, S3.for_sandbox() provides a factory helper that reads base storage environment settings and configures default local credentials:

from flyte.storage import S3

# Constructs S3 storage pointing to http://localhost:4566 with default sandbox credentials
sandbox_storage = S3.for_sandbox()

Under the hood, S3.for_sandbox() merges super()._auto_as_kwargs() with:

• endpoint: "http://localhost:4566"
• access_key_id: "minio"
• secret_access_key: "miniostorage"

Google Cloud Storage (GCS)

GCS manages Google Cloud Storage settings:

@dataclass(init=True, repr=True, eq=True, frozen=True)
class GCS(Storage):
    gsutil_parallelism: bool = False

Environment Variable	Dataclass Field	Default Value	Description
GCP_GSUTIL_PARALLELISM	gsutil_parallelism	False	Controls parallelism behavior for Google Cloud operations.

When using GCS.auto(), authentication relies on Google Cloud Application Default Credentials (ADC) sourced from the environment (such as GOOGLE_APPLICATION_CREDENTIALS or Workload Identity).

Azure Blob Storage (ABFS)

ABFS supports both storage account key authentication and Azure Active Directory (Service Principal / OAuth) credentials:

@dataclass(init=True, repr=True, eq=True, frozen=True)
class ABFS(Storage):
    account_name: typing.Optional[str] = None
    account_key: typing.Optional[str] = None
    tenant_id: typing.Optional[str] = None
    client_id: typing.Optional[str] = None
    client_secret: typing.Optional[str] = None

ABFS.auto() maps the following environment variables:

Environment Variable	Dataclass Field	Default Value	Description
AZURE_STORAGE_ACCOUNT_NAME	account_name	None	Azure storage account name.
AZURE_STORAGE_ACCOUNT_KEY	account_key	None	Secret key for storage account authentication.
AZURE_TENANT_ID	tenant_id	None	Azure Active Directory tenant ID.
AZURE_CLIENT_ID	client_id	None	Azure Active Directory application (client) ID.
AZURE_CLIENT_SECRET	client_secret	None	Azure Active Directory client secret.

Automatic Protocol Resolution at Runtime

In many scenarios (such as raw I/O operations with flyte.storage.get, flyte.storage.put, or loading dataframes via flyte.io.File), flyte-sdk dynamically constructs the necessary storage configuration without explicit initialization.

The get_configured_fsspec_kwargs helper in storage/_storage.py resolves URI protocols (s3://, gs://, abfs://, abfss://) at runtime:

# Extracted from storage/_storage.py
def get_configured_fsspec_kwargs(
    protocol: typing.Optional[str] = None, anonymous: bool = False
) -> typing.Dict[str, typing.Any]:
    if protocol:
        try:
            storage_config = get_storage()
        except InitializationError:
            storage_config = None

        match protocol:
            case "s3":
                from flyte.storage import S3

                if storage_config and isinstance(storage_config, S3):
                    return storage_config.get_fsspec_kwargs(anonymous=anonymous)
                return S3.auto().get_fsspec_kwargs(anonymous=anonymous)

            case "gs":
                from flyte.storage import GCS

                if storage_config and isinstance(storage_config, GCS):
                    return storage_config.get_fsspec_kwargs(anonymous=anonymous)
                return GCS.auto().get_fsspec_kwargs(anonymous=anonymous)

            case "abfs" | "abfss":
                from flyte.storage import ABFS

                if storage_config and isinstance(storage_config, ABFS):
                    return storage_config.get_fsspec_kwargs(anonymous=anonymous)
                return ABFS.auto().get_fsspec_kwargs(anonymous=anonymous)

If global storage was not initialized via flyte.init(), or if the requested URI protocol differs from the globally configured storage instance, Flyte falls back to S3.auto(), GCS.auto(), or ABFS.auto() to build filesystem arguments from the process environment.

Gotchas and Behavioral Details

• Empty String Variables Ignored: The set_if_exists helper evaluates isinstance(val, bool) or bool(val is not None and val). An environment variable set to an empty string (export FLYTE_AWS_ENDPOINT="") is treated as non-existent and will not override the dataclass field default.
• Base Variable Parsing in Subclasses: S3.auto() calls super()._auto_as_kwargs() and thus automatically captures UNION_STORAGE_DEBUG, UNION_STORAGE_RETRIES, and UNION_STORAGE_BACKOFF_SECONDS. GCS.auto() and ABFS.auto() construct their instances directly from their respective environment variables without calling _auto_as_kwargs(), preserving default retry and debug parameters unless overridden in code.
• Anonymous Access: When requesting anonymous read access (e.g., reading public S3 or ABFS datasets), get_configured_fsspec_kwargs(anonymous=True) instructs get_fsspec_kwargs to inject skip_signature=True into the underlying configuration dictionary.
• Execution Validation: When executing tasks in hybrid mode, _Runner._run_hybrid in _run.py validates that get_storage() returns an instance of S3, GCS, or ABFS, raising ValueError if an unsupported storage type is encountered.