Skip to main content

Managing Directories of Files

When handling collections of files across local workspaces and remote blob stores in flyte-sdk, passing directory paths as unstructured strings leads to missing files, broken cache semantics, and manual path translation. The flyte.io.Dir class solves this by providing a generic, strongly typed representation of directories with built-in async and sync inspection, iteration, and transfer utilities.

Creating and Uploading Directories​

To upload a local directory to remote storage and pass it between tasks, use the asynchronous Dir.from_local() class method.

from pathlib import Path
import pandas as pd
from flyte.io import Dir

# Create a local directory with sample data
local_dir = Path("/tmp/dataset")
local_dir.mkdir(parents=True, exist_ok=True)
pd.DataFrame({"a": [1, 2, 3]}).to_parquet(local_dir / "part-0.parquet")
pd.DataFrame({"a": [4, 5, 6]}).to_parquet(local_dir / "part-1.parquet")

# Upload the local directory to remote storage
remote_dir: Dir[pd.DataFrame] = await Dir[pd.DataFrame].from_local(
    local_path=local_dir,
    remote_path="s3://my-bucket/data/dataset/",
)

If you omit the remote_path argument, Dir.from_local() automatically generates a unique path in the configured default raw data storage:

# Upload with an auto-generated remote storage path
uploaded_dir = await Dir.from_local(local_path=local_dir)

Deterministic Cache Keys​

When passing a directory to discoverable or cached tasks, you can provide a precomputed hash key using dir_cache_key. This ensures Flyte computes predictable cache keys without inspecting the remote filesystem:

cached_dir = await Dir[pd.DataFrame].from_local(
    local_path=local_dir,
    remote_path="s3://my-bucket/data/dataset/",
    dir_cache_key="v1.0.0-hash-abc123",
)

Referencing Existing Remote Directories​

To reference a directory that already exists in remote blob storage without re-uploading, instantiate Dir with Dir.from_existing_remote():

existing_dir = Dir.from_existing_remote(
    remote_path="s3://my-bucket/data/dataset/",
    dir_cache_key="known-dataset-version",
)

Iterating and Walking Files​

The Dir class yields File[T] instances representing individual items inside the directory, allowing you to stream or inspect files without downloading the entire directory upfront.

Asynchronous Directory Traversal​

Use Dir.walk() to iterate over files asynchronously:

import pandas as pd
from flyte.io import Dir

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

# Asynchronously walk through all nested files recursively
async for file in dataset.walk():
    # 'file' is a File[pd.DataFrame] instance
    local_file_path = await file.download()
    print(f"Downloaded file to: {local_file_path}")

To limit directory traversal to immediate children without recursing into subdirectories, set recursive=False:

async for file in dataset.walk(recursive=False):
    print(file.path)

Synchronous Directory Traversal​

For local filesystems or synchronous contexts, iterate using Dir.walk_sync():

for file in dataset.walk_sync():
    local_file_path = file.download_sync()
    print(f"Sync downloaded: {local_file_path}")

Listing Files Non-Recursively​

To retrieve immediate files as a list, use list_files() or list_files_sync():

# Asynchronously list non-recursive immediate child files
files = await dataset.list_files()
for f in files:
    print(f.name, f.path)

# Synchronously list immediate child files
sync_files = dataset.list_files_sync()

Fetching Specific Files​

To look up a specific file by its relative name within the directory:

# Asynchronous lookup
csv_file = await dataset.get_file("metrics.csv")
if csv_file:
    print(f"Found file at: {csv_file.path}")

# Synchronous lookup
sync_csv_file = dataset.get_file_sync("metrics.csv")

Checking Directory Existence​

Check whether the directory exists in local or remote storage:

# Asynchronous check
if await dataset.exists():
    print("Remote directory is accessible")

# Synchronous check
if dataset.exists_sync():
    print("Directory exists")

Downloading Entire Directories​

Use Dir.download() to download all files from a directory into a local target path. If local_path is not specified, flyte-sdk creates a temporary directory prefixed with flyte-tmp-:

# Download to a specific local folder
downloaded_path = await dataset.download(local_path="/tmp/extracted_dataset/")

# Download to a generated temporary directory
temp_path = await dataset.download()

When calling download() on a directory that is already local, flyte-sdk avoids redundant copies if local_path is omitted or matches dataset.path.

Type Serialization and the Type Engine​

Dir integrates with Flyte's type system via flyte.io._dir.DirTransformer. Flyte represents Dir instances as multipart blob literals:

  • Blob Type: types_pb2.BlobType(dimensionality=BlobDimensionality.MULTIPART)
  • Format: Generic type T (e.g., Dir[pd.DataFrame]) sets format metadata on the blob.
  • Explicit I/O: DirTransformer transfers path URIs, format, and hash attributes into Flyte literals. It does not automatically upload or download directories during task execution. You control exactly when file transfers happen within your task functions.

Troubleshooting​

NotImplementedError on Sync Remote Operations​

Symptom: Calling Dir.from_local_sync() or Dir.download_sync() raises NotImplementedError("Sync upload/download is not implemented for remote paths").

Cause: Synchronous remote I/O is not supported for remote storage backends.

Solution: Use the async equivalents await Dir.from_local() and await dir.download() inside async task definitions:

# Incorrect
# dir_ref = Dir.from_local_sync("/tmp/local_dir", "s3://bucket/dir")

# Correct
dir_ref = await Dir.from_local("/tmp/local_dir", "s3://bucket/dir")

Pattern Filtering in walk_sync​

Symptom: Passing a glob pattern via walk_sync(file_pattern="*.csv") returns all files regardless of pattern matching.

Cause: walk_sync accepts file_pattern: str = "*" in its method signature, but the underlying filesystem walker iterates without applying glob filtering.

Solution: Filter files explicitly in Python when iterating:

import fnmatch

for file in dataset.walk_sync():
    if fnmatch.fnmatch(file.name, "*.csv"):
        # Process matching file
        pass