Working with Directories

Choose the directory reference that matches your workflow

When a task passes a directory to another task, flyte-sdk passes a Dir reference rather than eagerly downloading the directory. Import the public types from flyte.io:

from flyte.io import Dir, EmptyDir, File

Dir is a Pydantic model, generic in the type of its files, and implements SerializableType. Its persisted fields are path, name, format, and hash. path may be a local path or a remote URI. If you omit name, the pre_init validator derives it from Path(path).name; format defaults to "", and hash defaults to None.

Use a factory rather than constructing a Dir for ordinary directory creation:

• await Dir.from_local(path, ...) uploads or defers uploading a local directory in asynchronous code.
• Dir.from_local_sync(path, ...) is the synchronous counterpart.
• Dir.from_existing_remote(uri, dir_cache_key=...) creates a reference to data that already exists.
• Dir.new_remote(dir_name=..., hash=...) allocates a new remote path for output that you will populate.

For example, the basic asynchronous producer in examples/basics/dir.py creates a local directory and returns its Dir value:

@env.task
async def upload_directory_from_local() -> Dir:
    local_path = await create_test_local_directory()
    uploaded_dir = await Dir.from_local(local_path)
    print(f"Uploaded local directory {local_path} to remote: {uploaded_dir.path}")
    return uploaded_dir

To reference an existing object-store directory without copying it, pass its URI directly:

@env.task
async def create_reference_to_existing_remote(remote_path: str) -> Dir:
    dir_ref = Dir.from_existing_remote(remote_path)
    print(f"Created Dir reference to existing remote directory: {dir_ref.path}")
    return dir_ref

Dir.new_remote() obtains a generated path from internal_ctx().raw_data.get_random_remote_path(dir_name). It is the factory to use when the output location must be created before files are written. The Dir source documents that the constructor is primarily for references to existing directories.

Traverse and access the files

Directory operations have separate asynchronous and synchronous forms. The async methods are used without the _sync suffix; the synchronous methods have the suffix:

Operation	Asynchronous	Synchronous	Result
Traverse	async for file in d.walk(...)	for file in d.walk_sync(...)	File[T] iterator
List top-level files	await d.list_files()	d.list_files_sync()	list[File[T]]
Find a direct child	await d.get_file(name)	d.get_file_sync(name)	File[T] or None
Check existence	await d.exists()	d.exists_sync()	bool
Download the directory	await d.download(...)	d.download_sync(...)	local path as str

Traversal yields File objects, so you can read or download an individual file without first downloading the whole directory. The async example in examples/basics/dir_local.py reads matching files directly:

@env.task
async def process_dir(dir: Dir) -> dict[str, str]:
    file_contents = {}
    async for file in dir.walk(recursive=False):
        if file.name.endswith(".py"):
            async with file.open("rb") as f:
                content = bytes(await f.read())
                text_content = content.decode("utf-8")
                file_contents[file.name] = text_content[:100]
                print(f"Read {file.name}: {text_content[:50]}...")
    return file_contents

Use the synchronous APIs in an ordinary task:

@env.task
def read_files_in_directory_sync(d: Dir) -> dict[str, str]:
    file_contents = {}
    for file in d.walk_sync(recursive=False):
        if file.name.endswith(".txt"):
            with file.open_sync("rb") as f:
                file_contents[file.name] = f.read().decode("utf-8")
    return file_contents

The Dir implementation delegates traversal to the fsspec filesystem selected for self.path. walk() uses the async filesystem’s _walk when available and otherwise uses fs.walk; each result is converted to File[T](path=...). list_files() is implemented by collecting walk(recursive=False), while list_files_sync() returns list(self.walk_sync(recursive=False)).

recursive=False is the directory API’s top-level listing mode. In walk(), it forces max_depth = 2 before calling fsspec. You can also bound asynchronous traversal with max_depth:

@env.task
async def walk_directory_async(d: Dir) -> list[File]:
    all_files = []
    async for file in d.walk(recursive=True):
        print(f"  Found file: {file.name} at {file.path}")
        all_files.append(file)
    return all_files

For a known filename, get_file() and get_file_sync() build a path by joining the directory path and the supplied name. They check that direct child only; they do not recursively search. Both return None when the child does not exist:

@env.task
def get_specific_file_sync(d: Dir, file_name: str) -> File | None:
    file = d.get_file_sync(file_name)
    if file:
        print(f"Found file {file_name}: {file.path}")
        return file
    print(f"File {file_name} not found in directory {d.path}")
    return None

Download whole directories only when needed

Call download() or download_sync() when local processing requires the complete tree. With no destination, Dir asks flyte.storage for a random local path and appends the source directory name. For a remote path it calls the storage layer with recursive=True. For a local path, no copy is performed when no destination is supplied (or when the destination equals the source): the original path is returned.

Pass a destination when the contents should be copied into a particular directory. The implementation uses shutil.copytree(..., dirs_exist_ok=True) for local-to-local copies, and remote downloads also place the directory contents in the supplied destination:

@env.task
async def download_to_path(d: Dir) -> str:
    local_dir = await d.download("/tmp/my_data/")
    return local_dir

The same behavior is available synchronously as d.download_sync("/tmp/my_data/"). If only one object is required, prefer the yielded File and its download()/download_sync() method; the distributed random-forest example walks until it finds one parquet partition and downloads that file rather than the complete dataset.

What crosses a Flyte task boundary

DirTransformer is registered with the type engine when flyte.io._dir is imported. It represents a Dir as a Flyte multipart blob literal. to_literal() validates that the value is a Dir, takes python_val.path as the URI, and carries python_val.hash as the literal hash. If the value has a lazy uploader, the transformer invokes it before constructing the literal. The transformer performs serialization only; directory I/O remains in Dir and the storage/fsspec layers.

The inverse conversion requires a blob with multipart dimensionality. It reconstructs the expected Python type using the literal URI, metadata format, and optional hash. The round-trip behavior is exercised directly in tests/flyte/io_types/test_dirs.py:

f = Dir(path="s3://bucket/files/")
lt = TypeEngine.to_literal_type(Dir)
lv = await DirTransformer().to_literal(f, Dir, lt)
pv = await DirTransformer().to_python_value(lv, Dir)
assert pv == f

A cache key supplied through dir_cache_key is preserved in the literal and on the reconstructed Dir:

d = Dir.from_existing_remote("s3://bucket/data/", dir_cache_key="abc123")
lt = TypeEngine.to_literal_type(Dir)
lv = await DirTransformer().to_literal(d, Dir, lt)
pv = await DirTransformer().to_python_value(lv, Dir)
assert lv.hash == "abc123"
assert pv.path == d.path
assert pv.hash == "abc123"

Dir.Config adds the generated JSON-schema metadata description: "A directory reference with an optional format type." and x-flyte-type: "dir", while allowing arbitrary types. The transformer’s literal type currently uses multipart dimensionality and an empty literal format; the source has a TODO to derive that format from the generic type. A non-blob literal or a blob that is not multipart is rejected by DirTransformer with TypeTransformerFailedError.

Deferred local-directory upload

from_local() and from_local_sync() normally resolve a destination through the active raw-data context. When there is no raw-data context and no explicit destination, they instead return a Dir retaining the local path and attach a private lazy uploader. During literal conversion, DirTransformer.to_literal() invokes that uploader.

The uploader checks _get_main_run_mode(): in local mode it returns the original local path; otherwise it calls flyte.remote.upload_dir.aio. Consequently, await Dir.from_local(...) does not always mean that the returned Dir.path is already an object-store URI. The local-mode behavior is covered by tests/flyte/test_run_mode.py, which passes a locally created Dir into a task and successfully walks its local files.

For asynchronous uploads with a raw-data context, Dir.from_local() calls storage.put(..., recursive=True, batch_size=batch_size). Its batch_size argument controls upload concurrency; when omitted, the source documentation says the default is determined by FLYTE_IO_BATCH_SIZE, whose documented default is 32. The synchronous factory has no batch_size parameter and calls the selected filesystem’s put(..., recursive=True).

Represent an optional directory output

Do not use Optional[Dir] for a directory field that must cross Flyte’s dataclass/type-engine boundary. PythonScriptOutput demonstrates the supported shape: its output_dir field is always typed as flyte.io.Dir. When a script has no output directory, _run_python_script.py assigns EmptyDir() instead:

@dataclass
class PythonScriptOutput:
    exit_code: int
    stdout: str
    output_dir: flyte.io.Dir

The production task constructs the value as follows:

from flyte.io import Dir, EmptyDir

if output_dir:
    _dir: Dir = await Dir.from_local(output_dir)
else:
    _dir = EmptyDir()

Dir.empty() is the equivalent convenience factory:

empty_dir = Dir.empty()
assert empty_dir.is_empty

EmptyDir is a Dir subclass whose initializer forcibly replaces every supplied path with the sentinel "::flyte-empty-dir::" and defaults name to "". Dir.is_empty checks for that sentinel path. On deserialization, DirTransformer.to_python_value() constructs the expected Python type, so the receiving value is a plain Dir, not an EmptyDir instance. Branch on the property rather than on subclass identity:

@env.task
async def consume_output(output: Dir) -> str:
    if output.is_empty:
        return "No directory was produced"
    return str(len(await output.list_files()))

The sentinel must be checked before filesystem operations. Its path is not a directory to walk, download, or test; it is the marker used to carry “no directory was produced.” The reason for this representation is stated in EmptyDir and PythonScriptOutput: mashumaro’s DataclassTransformer strips Optional from SerializableType fields and can call Dir._deserialize(None), which fails. Keeping the field as plain Dir allows the sentinel path to round-trip.

Operational caveats

• Match the API to the task style: consume walk() with async for and use File.open()/File.download() asynchronously; consume _sync iterators with for and use the corresponding synchronous File methods.
• walk_sync() currently accepts recursive, file_pattern, and max_depth, but its implementation calls fs.walk(self.path, maxdepth=max_depth) without applying recursive or file_pattern. In particular, the advertised file_pattern="*.txt" example does not currently filter yielded files. Filter by file.name yourself when using the synchronous iterator.
• get_file() uses the filesystem separator and calls fs.exists() directly; it is a direct-child lookup, not a recursive or content search.
• Configure Flyte’s raw-data and storage context before relying on generated remote destinations. from_existing_remote() avoids an upload and simply stores the URI you provide.
• The multipart representation is part of the type contract. DirTransformer.guess_python_type() returns Dir only for multipart blob types and raises ValueError for other literal types.
• No dedicated focused tests for EmptyDir, Dir.empty(), or is_empty were found; their behavior is defined by the implementation, docstrings, and the production PythonScriptOutput path.