Working with Files and Directories

Files and directories are references to storage locations

When a task receives File[DataFrame] or Dir[DataFrame], the value identifies a location; it is not the file or directory contents. Construct the reference with a local path or a URI understood by flyte.storage:

from pandas import DataFrame
from flyte.io import Dir, File

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

File and Dir are Pydantic models and SerializableType implementations in io/_file.py and io/_dir.py. Both have path, optional name, format, and optional hash fields. If name is omitted, their pre_init validator sets it to Path(data["path"]).name. Consequently, the path is the authoritative reference, while name, format, and hash are metadata carried with it. The generic parameter T—for example, DataFrame—documents the expected file format/type; the transformer currently leaves the literal format empty when it constructs a declared type.

The classes do not automatically read, upload, or download data merely because they are instantiated or passed through a task. Their docstrings explicitly assign I/O to the user. Choose an operation that matches the lifecycle of the reference:

• Use File.from_existing_remote or Dir.from_existing_remote when the storage location already exists.
• Use File.from_local or Dir.from_local when a local source must be transferred.
• Use File.new_remote when a task will write a new file directly to a generated remote location.

Reference an existing location

For an existing remote object, construct a reference without a storage transfer:

from flyte.io import Dir, File

remote_file = File.from_existing_remote("s3://my-bucket/data.csv")
remote_dir = Dir.from_existing_remote("s3://my-bucket/data/")

# Supply a known cache identity when one is available.
remote_file_with_key = File.from_existing_remote(
    "s3://my-bucket/data.csv", file_cache_key="abc123"
)
remote_dir_with_key = Dir.from_existing_remote(
    "s3://my-bucket/data/", dir_cache_key="abc123"
)

File.from_existing_remote stores remote_path and the optional file_cache_key as path and hash; it does not verify that the URI exists. Dir.from_existing_remote behaves similarly with dir_cache_key. If no key is supplied, the file docstring says that discovery hashing is based on the object's attributes—principally its path—not remote contents.

The same reference model is used for local and remote paths. For example, the CLI parameter types validate local input before constructing a reference: DirParamType requires a local value to exist and be a directory, while FileParamType requires a local value to exist and be a file. Remote values bypass those local filesystem checks; the file parameter uses File.from_existing_remote(value), and the directory parameter uses Dir(path=value) (cli/_params.py).

Read and write a File

Use File.open as an asynchronous context manager when the surrounding code is async:

from pandas import DataFrame
from flyte.io import File

async def read_csv_file(file: File[DataFrame]):
    async with file.open() as f:
        data = await f.read()
    return data

The default mode is "rb". File.open asks storage.get_underlying_filesystem(path=self.path) which filesystem handles the path. For a local filesystem (fs.protocol == "file"), it uses aiofiles.open. For other protocols, it requires binary mode and raises ValueError if "b" is absent:

async with remote_file.open("r") as f:
    data = await f.read()  # ValueError: remote mode must include "b"

For a remote asynchronous fsspec filesystem, File.open calls open_async when available. If that operation raises NotImplementedError, it falls back to synchronous fs.open; this fallback can perform synchronous I/O inside the async method. The method also accepts block_size, cache_type, cache_options, compression, and additional keyword arguments. Caching is applied unless cache_type="none".

Use open_sync for synchronous access:

from flyte.io import File

local_file = File(path="/tmp/data.csv")
with local_file.open_sync("rb") as f:
    data = f.read()

open_sync always obtains the filesystem and calls fs.open, forwarding compression, block size, cache settings, and extra arguments. File.exists_sync() uses the same selected filesystem:

if local_file.exists_sync():
    with local_file.open_sync() as f:
        data = f.read()

There is no synchronous File.download method. The implemented download API is asynchronous:

local_path = await remote_file.download("/tmp/data.csv")

With no destination, File.download asks storage for a random local path. For local files it copies with aiofiles; for non-local protocols it calls storage.get(self.path, local_path).

Stream a new remote file

Inside initialized Flyte execution, File.new_remote() creates a reference whose path comes from internal_ctx().raw_data.get_random_remote_path(). Open that reference for writing and return it:

from pandas import DataFrame
from flyte.io import File

async def write_csv_file() -> File[DataFrame]:
    file = File.new_remote()
    async with file.open("wb") as f:
        f.write(b"value\n1\n")
    return file

new_remote is decorated with @requires_initialization, so Flyte must be initialized before it is called. A string passed as hash_method is stored as a known cache value; a HashMethod instance is stored as the hashing method.

The remote fallback in File.open wraps the handle with HashingWriter only when hash_method is set and hash is still unset. After the context yields and the writer completes, its result is assigned to self.hash. The async open_async branch does not receive that wrapper, so hashing behavior differs between the two remote-open paths.

Upload a local file

File.from_local is asynchronous and initialization-gated. It first checks os.path.exists(local_path) and raises ValueError when the source does not exist. Pass a file path and await the call:

from flyte.io import File

async def upload_file() -> File:
    return await File.from_local(
        "/tmp/data.csv",
        remote_destination="s3://my-bucket/data.csv",
    )

If remote_destination is omitted, the method obtains a generated remote path from internal_ctx().raw_data. A local destination with no explicit destination is optimized to reference the absolute source path without copying. An explicit local destination copies the bytes. A non-local destination uses storage.put; when a HashMethod is supplied, the implementation may use AsyncHashingReader and storage.put_stream, while PrecomputedValue supplies its result without streaming through the hashing reader.

The source validates existence, not regular-file type, so callers should provide a file rather than a directory. Also note that a few examples retained in the File docstring omit await; the method is declared async, and calling it requires awaiting it.

Traverse a Dir

A directory produces File[T] references during traversal; it does not yield file contents. The async-first pattern is:

from pandas import DataFrame
from flyte.io import Dir

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

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

Dir.walk selects an fsspec filesystem from self.path. It uses the filesystem's async _walk for an AsyncFileSystem and synchronous walk otherwise, converting returned paths back to protocol-aware paths before constructing File[T](path=full_file). Recursive walking is the default. Passing recursive=False forces max_depth=2:

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

For synchronous traversal, use walk_sync and File.open_sync:

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

Although walk_sync accepts recursive and file_pattern, its implementation currently calls fs.walk(self.path, maxdepth=max_depth) and does not use file_pattern. It also does not apply the async method's recursive=False depth override. Use max_depth explicitly when you need to constrain synchronous traversal.

For a non-recursive list, use the convenience methods:

files = await data_dir.list_files()
files_sync = data_dir.list_files_sync()

list_files collects walk(recursive=False), and list_files_sync returns list(self.walk_sync(recursive=False)). To look up one child by name, use either variant:

file = await data_dir.get_file("data.csv")
if file is not None:
    async with file.open() as f:
        content = await f.read()

file_sync = data_dir.get_file_sync("data.csv")
if file_sync is not None:
    with file_sync.open_sync() as f:
        content = f.read()

get_file joins the directory path with the filesystem separator and returns None when fs.exists is false. get_file_sync uses os.path.join and File.exists_sync. The async method calls fs.exists directly even when the selected filesystem is asynchronous, so it is not equivalent to Dir.exists, which calls _exists for an AsyncFileSystem.

Upload and download a directory

Upload a local directory recursively with Dir.from_local:

from flyte.io import Dir

async def upload_directory() -> Dir:
    return await Dir.from_local(
        "/tmp/data_dir/",
        remote_path="s3://my-bucket/data/",
        dir_cache_key="abc123",
    )

The method calls storage.put(..., recursive=True), derives the directory name from the normalized local path, and returns a Dir containing the resulting path, name, and optional hash. Dir.from_local_sync exists as a method but always raises NotImplementedError; synchronous directory upload is not implemented.

For asynchronous downloads, Dir.download accepts an optional local destination. It uses storage.get(..., recursive=True) for remote directories. For a local directory, no destination—or the same path—returns the existing path; a different destination is copied with shutil.copytree in an executor before the method returns. Dir.download_sync copies local directories, but raises NotImplementedError("Sync download is not implemented for remote paths") for remote directories:

local_dir = await data_dir.download("/tmp/my_data/")
local_dir_sync = local_data_dir.download_sync("/tmp/my_data-copy/")

Local and remote behavior at a glance

Operation	Local path	Remote path
File.open	Uses aiofiles	Uses async fsspec opening when available, then synchronous fs.open fallback; mode must contain b
File.open_sync	Uses fs.open	Uses fs.open
File.download	Copies with aiofiles	Calls storage.get asynchronously
Dir.walk	Uses the selected filesystem's walk implementation	Uses the selected filesystem's async or sync fsspec walk implementation
Dir.download	Returns the same path or copies locally	Calls storage.get(..., recursive=True)
Dir.download_sync	Copies with shutil.copytree	Raises NotImplementedError

Protocol selection is centralized through flyte.storage: File and Dir do not contain provider-specific S3, GCS, or Azure transfer code. The configured storage layer resolves the filesystem and transfer operations for the URI protocol.

Flyte serialization and container integration

Importing File and Dir registers FileTransformer and DirTransformer with TypeEngine (io/_file.py and io/_dir.py). The transformers carry references, not bytes:

• FileTransformer serializes a File as a single-part Flyte blob whose URI is python_val.path, whose metadata format is python_val.format, and whose literal hash is python_val.hash when present.
• DirTransformer performs the corresponding operation as a multipart blob.
• Deserialization reconstructs a File or Dir from the blob URI, metadata format, and optional literal hash.
• FileTransformer rejects non-blob and non-single-part literals; DirTransformer rejects non-blob and non-multipart literals.

The transformer get_literal_type methods currently use format="" and contain a TODO to derive format from the generic type. schema_match on both models compares the generated JSON-schema type, title, and required-field set; the type engine uses that shape when recognizing these objects in reconstructed fields and list elements.

ContainerTask uses the reference's .path for file and directory inputs. Its _render_command_and_volume_binding binds that path to /var/inputs/<input-name> and requires path-like command syntax rather than {{.inputs.name}} for these values:

/var/inputs/infile

Using template syntax for a File or Dir causes the container integration to raise an assertion with the requested path-like form. After a container runs, _convert_output_val_to_correct_type calls await File.from_local(output_path) for File outputs and await Dir.from_local(output_path) for Dir outputs, transferring the generated local outputs through the configured storage layer.

Initialization, storage, and cache metadata

Initialize Flyte before using File.new_remote or File.from_local; both methods are decorated with @requires_initialization and use the initialized raw-data context to generate or resolve destinations. Storage provider behavior is selected by URI protocol and its configuration, including the storage debug/retry settings and provider credentials such as FLYTE_AWS_ACCESS_KEY_ID, FLYTE_AWS_SECRET_ACCESS_KEY, Azure storage settings, and the optional FLYTE_AWS_ENDPOINT.

Use hash when you already have a cache identity, or use the hashing options supported by File.from_local and File.new_remote when the content should contribute to it. Dir.from_local accepts dir_cache_key and stores it directly as hash. These values are metadata on the reference and are serialized into the Flyte literal; they do not change the fact that FileTransformer and DirTransformer transport a URI rather than file contents.