Processing Directory Contents

Process the contents of a Dir

To process every file represented by a Dir[T], traverse the directory with walk() or walk_sync() and then perform I/O explicitly on each File[T] that it yields:

from pandas import DataFrame
from flyte.io import Dir


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


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

Dir is a typed reference containing a local path or remote URI. walk() does not return file contents or strings: it resolves the filesystem for data_dir.path and yields new File[T](path=...) objects. The File object is the point at which reading, writing, or downloading occurs.

Walk recursively or non-recursively

walk() is an asynchronous iterator and defaults to recursive traversal. Pass recursive=False when you want the non-recursive form exposed by the API:

from pandas import DataFrame
from flyte.io import Dir


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


async def process_immediate_files():
    async for file in data_dir.walk(recursive=False):
        async with file.open("rb") as f:
            data = await f.read()
            # Process data

The method obtains an fsspec filesystem through storage.get_underlying_filesystem(path=self.path). For an asynchronous filesystem it uses fs._walk; otherwise it uses fs.walk. Local/file protocols reconstruct paths with os.path.join, while other protocols use the filesystem separator and unstrip_protocol. Treat the yielded .path values as backend paths rather than constructing equivalent paths manually.

You can also bound recursive traversal with max_depth:

async for file in data_dir.walk(max_depth=2):
    async with file.open() as f:
        content = await f.read()

When recursive=False, walk() overwrites max_depth with 2. The exact result depends on the depth semantics of the selected fsspec backend.

Use synchronous traversal

For synchronous processing, walk_sync() yields the same kind of File[T] references and pairs with File.open_sync():

from pandas import DataFrame
from flyte.io import Dir


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


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

The two traversal methods differ as follows:

Method	Result	Default	Depth and backend behavior
walk()	Async iterator of File[T]	recursive=True	Uses the filesystem's async walk when available; recursive=False sets max_depth=2.
walk_sync()	Synchronous iterator of File[T]	recursive=True	Calls the synchronous filesystem walk with the supplied max_depth.

walk_sync() declares file_pattern="*", but its implementation never uses that argument. For example, this call does not currently filter the yielded files:

for file in data_dir.walk_sync(file_pattern="*.csv"):
    with file.open_sync() as f:
        content = f.read()

Also note that walk_sync() does not translate recursive=False into a max_depth value. list_files_sync() calls walk_sync(recursive=False), but the synchronous walker receives max_depth=None. Consequently, do not assume that synchronous non-recursive listing has strict immediate-child behavior for every filesystem backend; verify it with the filesystem used by the Dir.

Materialize a file list

Use list_files() when the processing code needs a list rather than a streaming iterator:

from pandas import DataFrame
from flyte.io import Dir


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


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

list_files() creates an empty list, consumes walk(recursive=False), and appends each yielded File[T]. Its synchronous counterpart materializes walk_sync(recursive=False):

from pandas import DataFrame
from flyte.io import Dir


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


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

Method	Result	Implementation
list_files()	List[File[T]]	Collects walk(recursive=False) asynchronously.
list_files_sync()	List[File[T]]	Returns list(walk_sync(recursive=False)).

The list methods only materialize references; they do not open or download the files. Each yielded File is constructed with its discovered path, so Dir.format, Dir.hash, and other directory metadata are not copied to the individual file references.

Process, download, or pass each File

File.open() and File.open_sync() are context managers over the selected filesystem. For local files, asynchronous opening uses aiofiles; remote asynchronous opening uses an async fsspec filesystem when available and otherwise falls back to synchronous fsspec opening. Remote asynchronous access requires a binary mode containing b:

async for file in data_dir.walk():
    async with file.open("rb") as f:
        data = await f.read()

For a remote file, a mode such as "r" raises ValueError("Mode must include 'b' for binary access, when using remote files."). The synchronous context manager delegates to fs.open() and accepts the mode and fsspec options defined by File.open_sync().

If processing requires a local copy, use the asynchronous file download explicitly:

async for file in data_dir.walk():
    local_file = await file.download()
    # Process local_file

The Dir and File transformers describe references for Flyte's type system; they do not automatically perform the traversal, open, upload, or download. DirTransformer registers a multipart blob whose URI is Dir.path, while FileTransformer registers the corresponding single-file reference. Traversal and file I/O remain caller operations.

Use directory references with container tasks

Container integration binds a File or Dir input's .path into the configured container input directory. The command must use path-like syntax, such as /var/inputs/infile; template syntax for these types raises an assertion:

def _render_command_and_volume_binding(self, cmd: str, **kwargs):
    from flyte.io import Dir, File

    volume_binding = {}
    path_k = _extract_path_command_key(cmd, str(self._input_data_dir))
    keys = [path_k] if path_k else _extract_command_key(cmd)

    command = cmd
    if keys:
        for k in keys:
            input_val = kwargs.get(k)
            if input_val and type(input_val) in [File, Dir]:
                if not path_k:
                    raise AssertionError(
                        "File and Directory commands should not use the template syntax "
                        "like this: {{.inputs.infile}}\n"
                        "Please use a path-like syntax, such as: /var/inputs/infile."
                    )
                local_flyte_file_or_dir_path = input_val.path
                remote_flyte_file_or_dir_path = os.path.join(self._input_data_dir, k)
                volume_binding[local_flyte_file_or_dir_path] = {
                    "bind": remote_flyte_file_or_dir_path,
                    "mode": "rw",
                }
    return command, volume_binding

This is the behavior in extras/_container.py: exact File and Dir types are recognized, and their stored paths become the source side of the volume binding. Container outputs declared as File or Dir are converted with File.from_local(output_path) or Dir.from_local(output_path), respectively.

Create the directory reference

For an existing remote directory, construct a reference without transferring data:

from flyte.io import Dir


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

To upload a local directory recursively, use the asynchronous factory:

from pandas import DataFrame
from flyte.io import Dir


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

Dir.from_local() calls storage.put(..., recursive=True) and returns a Dir pointing at the resulting path. If remote_path is omitted, remote-path generation depends on the initialized execution context's raw-data configuration. Dir.from_local_sync() is not implemented and raises NotImplementedError("Sync upload is not implemented for remote paths").

For local directories, Dir.download() can return the original path or copy/download recursively to a selected local destination. Dir.download_sync() can copy a local directory, but raises NotImplementedError("Sync download is not implemented for remote paths") for remote directories.

Troubleshoot backend and listing behavior

• The directory cannot be walked: Dir.walk() and walk_sync() select an fsspec filesystem from Dir.path. Ensure the URI protocol and its storage configuration are available. S3-specific settings are read by S3.auto() from FLYTE_AWS_ENDPOINT, FLYTE_AWS_ACCESS_KEY_ID, and FLYTE_AWS_SECRET_ACCESS_KEY; general storage settings include UNION_STORAGE_DEBUG, UNION_STORAGE_RETRIES, and UNION_STORAGE_BACKOFF_SECONDS.
• A sync glob does not filter: file_pattern is present in walk_sync()'s signature and documentation but is not applied by the implementation. Filter the returned File objects in caller code.
• Sync listing includes more than expected: list_files_sync() passes recursive=False, but walk_sync() forwards max_depth=None. Check the selected filesystem's walk behavior instead of relying on the method name.
• Generated output paths fail before execution: File.new_remote() and destination-less File.from_local() depend on internal_ctx().raw_data; File.new_remote() also requires Flyte initialization through its requires_initialization decorator. A destination-less Dir.from_local() likewise relies on the configured raw-data path.
• A remote async open rejects text mode: include "b" in the mode, for example file.open("rb"). Local files use the local aiofiles path and do not take that remote-mode check.

The implementation docstrings in io/_dir.py and io/_file.py provide the repository's traversal examples; no matching repository tests or separate example files were found for this workflow. Validate depth and path reconstruction against the target storage backend when those details matter.