Interacting with Directories

Work with directory references

If a task receives a directory URI such as s3://my-bucket/data/, use flyte.io.Dir as the directory-level reference and choose an async or sync API for consuming its files:

from pandas import DataFrame
from flyte.io import Dir


data_dir = Dir[DataFrame](path="s3://my-bucket/data/")


async def read_directory():
    async for file in data_dir.walk():
        async with file.open() as f:
            content = await f.read()
            # Process content


for file in data_dir.walk_sync():
    with file.open_sync() as f:
        content = f.read()
        # Process content

Dir is a generic Pydantic model. Its path is either a local or remote path; name is optional; format defaults to an empty string; and hash is optional. When name is omitted, Dir derives it from Path(path).name. The T in Dir[T] describes the format of files yielded as File[T]; it does not make the directory transformer perform file I/O.

For example, these two references have the same directory path but explicitly carry different metadata:

from flyte.io import Dir


data_dir = Dir(path="/tmp/data_dir", format="csv", hash="abc123")
assert data_dir.name == "data_dir"

The public import is from flyte.io import Dir; flyte.io re-exports the class from its internal directory module.

Iterate through files

Use walk() when the surrounding code is asynchronous. It yields File[T] objects, not local path strings, so use the File API to open or download each item:

from flyte.io import Dir


directory = Dir(path="s3://my-bucket/data/")


async def download_files():
    async for file in directory.walk():
        local_path = await file.download()
        # Process the file at local_path

walk() obtains a filesystem from flyte.storage.get_underlying_filesystem(path=self.path). For an fsspec.asyn.AsyncFileSystem, it consumes the filesystem's _walk; otherwise it uses the filesystem's synchronous walk. The yielded File paths are reconstructed with the filesystem's protocol handling, so the File retains the directory's local or remote path form.

Traversal is recursive by default. To request the non-recursive form, use recursive=False or use list_files():

from flyte.io import Dir


directory = Dir(path="s3://my-bucket/data/")


async def immediate_files():
    files = await directory.list_files()
    for file in files:
        async with file.open() as stream:
            content = await stream.read()
            # Process content

list_files() consumes walk(recursive=False) and returns a list. In the async implementation, recursive=False is implemented by setting max_depth=2 before calling the filesystem walker. That behavior depends on the depth semantics of the selected fsspec provider; it is not a separate direct-child listing operation.

Synchronous traversal

The synchronous equivalents return ordinary iterators and lists:

from flyte.io import Dir


directory = Dir(path="/tmp/data_dir")

for file in directory.walk_sync():
    local_path = file.download_sync()
    # Process the file

files = directory.list_files_sync()

list_files_sync() is implemented as list(self.walk_sync(recursive=False)). There are two details to account for when switching from async to sync traversal:

• walk_sync() accepts file_pattern="*", but the implementation never applies that argument. Passing a pattern therefore does not filter the yielded files.
• Although list_files_sync() passes recursive=False, walk_sync() does not use its recursive parameter to force a depth. It passes max_depth directly to the fsspec walk, unlike walk()'s recursive=False path. Do not assume the two listing methods have identical depth behavior.

Download an entire directory

Use the asynchronous download() method when the directory is remote or when the caller wants an asynchronous transfer:

from flyte.io import Dir


directory = Dir.from_existing_remote("s3://bucket/data/")


async def download_directory():
    local_dir = await directory.download("/tmp/my_data/")
    return local_dir

download() chooses a temporary local destination with storage.get_random_local_path() when local_path is omitted. For a remote source, it calls storage.get(self.path, local_dest, recursive=True), so the directory transfer is recursive.

For a local source, download() returns the original path without copying when no destination is supplied or when the supplied destination equals self.path. If a different destination is supplied, it copies the tree with shutil.copytree(..., dirs_exist_ok=True) in an executor before completing the storage operation.

The sync method only supports the local cases:

from flyte.io import Dir


local_directory = Dir(path="/tmp/data_dir")

same_path = local_directory.download_sync()
copy_path = local_directory.download_sync("/tmp/data_copy")

download_sync() returns the source path when no local copy is needed and uses shutil.copytree(..., dirs_exist_ok=True) for a different local destination. For a remote directory it raises:

NotImplementedError: Sync download is not implemented for remote paths

Use await directory.download(...) for remote directories rather than relying on download_sync().

Create directory outputs from local data

To upload local directory data and obtain a Dir reference, use the asynchronous from_local() factory:

from pandas import DataFrame
from flyte.io import Dir


async def create_output():
    remote_dir = await Dir[DataFrame].from_local(
        "/tmp/data_dir/",
        "s3://bucket/data/",
    )
    return remote_dir

from_local() converts the local path to a string, derives the name from the normalized local path, and calls flyte.storage.put(..., recursive=True). The returned Dir points to the resulting upload path. A precomputed cache key can be stored in hash:

from flyte.io import Dir


async def create_cached_output():
    return await Dir.from_local(
        "/tmp/data_dir/",
        "s3://bucket/data/",
        dir_cache_key="abc123",
    )

from_local() does not first validate that local_path is a directory; it delegates to storage. If remote_path is omitted or remote, the configured or inferred storage provider must be available.

When the remote directory already exists, create a reference without uploading or checking existence:

from flyte.io import Dir


remote_dir = Dir.from_existing_remote("s3://bucket/data/")
remote_dir_with_hash = Dir.from_existing_remote("s3://bucket/data/", dir_cache_key="abc123")

Call exists() or exists_sync() separately when the reference must be checked. from_local_sync() is present as an API, but always raises NotImplementedError("Sync upload is not implemented for remote paths"); it is not a synchronous alternative to from_local().

Check existence and select one file

Both async and sync existence helpers resolve the filesystem from the directory path:

from flyte.io import Dir


directory = Dir(path="s3://bucket/data/")


async def find_data():
    if await directory.exists():
        return await directory.get_file("data.csv")
    return None


def find_local_data():
    local_directory = Dir(path="/tmp/data_dir")
    if local_directory.exists_sync():
        return local_directory.get_file_sync("data.csv")
    return None

exists() calls _exists for an AsyncFileSystem and exists for other filesystems. exists_sync() calls the filesystem's synchronous exists method. get_file() joins the directory path with the requested name using the filesystem separator and returns a File[T] only if that path exists. get_file_sync() uses os.path.join and File.exists_sync() instead. Consequently, remote path joining can differ between the two lookup methods; use the method matching the execution model and provider.

Use Dir in container tasks

ContainerTask treats an input whose exact type is File or Dir as a path-like mounted input. The command must contain the input-data path rather than a template expression. The relevant command shape is:

cmd = "/var/inputs/infile"

For such an input, container handling uses input_val.path as the host-side path and binds it to the corresponding path under the container input data directory. A template such as {{.inputs.infile}} raises an assertion for File and Dir inputs; the container code explicitly requires a path-like command such as /var/inputs/infile.

For container outputs, when the declared output type is Dir, ContainerTask calls await Dir.from_local(output_path). Thus a directory written to the local container output path becomes an asynchronously uploaded Dir reference before task output serialization.

Understand Flyte literal conversion

DirTransformer is defined in flyte.io._dir and registered at module import with TypeEngine.register(DirTransformer()). It represents a directory as a Flyte multipart blob, but it does not upload or download anything:

from flyte.io import Dir


directory = Dir(path="s3://bucket/data/", format="csv", hash="abc123")

get_literal_type() returns a BlobType with MULTIPART dimensionality and an empty format. When a concrete Dir is converted by to_literal(), the transformer writes the value's path to the blob URI, its format to blob metadata, and its optional hash to the Flyte literal. It rejects values that are not Dir instances.

On the way back, to_python_value() requires a blob literal whose dimensionality is MULTIPART; otherwise it raises TypeTransformerFailedError. It reconstructs a Dir from the blob URI, metadata format, and optional literal hash. guess_python_type() maps multipart blob literal types to Dir and raises ValueError for other literal types. The runtime's TypeEngine conversion path therefore serializes the reference and metadata, while directory data movement remains the responsibility of Dir methods or the surrounding task integration.

Configure remote directory access

Directory walking, existence checks, and transfers use fsspec through flyte.storage. The URI selects the underlying provider; the storage configuration must provide the provider support and credentials for paths such as s3://, gs://, or abfs://.

For S3-backed directories, the storage configuration reads FLYTE_AWS_ACCESS_KEY_ID and FLYTE_AWS_SECRET_ACCESS_KEY; FLYTE_AWS_ENDPOINT can select an S3-compatible endpoint. Azure-backed abfs:// operations use the configured Azure variables, including AZURE_STORAGE_ACCOUNT_NAME, AZURE_STORAGE_ACCOUNT_KEY, and the service-principal variables. GCS provider behavior includes the GCP_GSUTIL_PARALLELISM setting. Storage-wide behavior also includes UNION_STORAGE_DEBUG, UNION_STORAGE_RETRIES, and UNION_STORAGE_BACKOFF_SECONDS.

Troubleshooting checklist

• A remote download_sync() fails with NotImplementedError: use async download().
• A synchronous upload fails immediately: from_local_sync() is intentionally unimplemented; use async from_local().
• A walk_sync(file_pattern="*.csv") still yields other files: file_pattern is accepted but unused.
• Async and sync non-recursive listing returns different depths: async walk(recursive=False) forces max_depth=2, while sync walking passes its max_depth unchanged.
• A remote reference does not exist: from_existing_remote() performs no existence check; call await exists() or exists_sync().
• A container directory input is rejected: use a path-like command such as /var/inputs/infile, not {{.inputs.infile}}.
• A remote operation cannot resolve its filesystem: verify the URI protocol, provider configuration, and credentials used by flyte.storage.