Working with Files

Choose the way you want to materialize the file

File represents one file path, not the file's contents. Import the public type from flyte.io:

from flyte.io import File

Use the factory that matches the file's lifecycle:

Situation	API	What it does
You will produce bytes directly in storage	File.new_remote() followed by open("wb") or open_sync("wb")	Allocates a generated raw-data path and returns a writable reference.
You already have a local file	await File.from_local(path) or File.from_local_sync(path)	Copies or uploads the file, unless the current context can keep it local or defer the upload.
A URI already exists	File.from_existing_remote(uri)	Creates only a reference; it does not upload or copy.
A produced artifact must have a stable name	File.named_remote(name)	Derives a deterministic path from the active task context.

The model has path, optional name, format, and optional hash fields. If name is omitted, the validator in src/flyte/io/_file.py derives it from Path(path).name. hash_method is retained for hashing transfers and is excluded from the serialized model. Direct construction such as File(path=...) is therefore useful for an existing reference, but it does not move bytes.

Before using context-sensitive factories, initialize Flyte. new_remote, named_remote, and the local-upload factories are guarded by requires_initialization; generated paths come from the active raw-data configuration.

Write and read asynchronously

For an asynchronous task, create a remote reference and use it as an async context manager. The handle returned by open supports asynchronous reads and writes:

import flyte
from flyte.io import File

env = flyte.TaskEnvironment("file")

@env.task
async def create_new_remote_file(content: str) -> File:
    f = File.new_remote()
    async with f.open("wb") as fh:
        await fh.write(content.encode("utf-8"))
    return f

@env.task
async def read_file_async(f: File) -> str:
    async with f.open("rb") as fh:
        contents = bytes(await fh.read())
        return contents.decode("utf-8")

The bytes(...) conversion is significant for the storage path used by flyte-sdk: an asynchronous read() can return a memoryview-like value. open defaults to binary read mode ("rb"). Pass "rt" or another supported fsspec mode when you explicitly want text handling.

For large reads, pass block_size and consume chunks until an empty chunk is returned:

@env.task
async def stream_read(f: File) -> str:
    content_parts = []
    async with f.open("rb", block_size=1024) as fh:
        while True:
            chunk = await fh.read()
            if not chunk:
                break
            content_parts.append(chunk)
    return b"".join(content_parts).decode("utf-8")

File.open forwards mode, block_size, cache_type, cache_options, compression, and additional keyword arguments to flyte.storage.open. Its default cache type is "readahead"; the documented choices also include "mmap", "bytes", and "none". The context manager closes the returned handle and awaits the close operation when the handle provides an awaitable close.

If asynchronous storage is unsupported, open falls back to aiofiles.open only when the underlying protocol is local ("file"). For a non-local protocol, the original OnlyAsyncIOSupportedError is re-raised.

Use the synchronous interface

Use the _sync methods in a synchronous task. The returned handle follows the normal with and .read()/.write() conventions:

import flyte
from flyte.io import File

env = flyte.TaskEnvironment("sync_file")

@env.task

def create_new_remote_file(content: str) -> File:
    f = File.new_remote()
    with f.open_sync("wb") as fh:
        fh.write(content.encode("utf-8"))
    return f

@env.task

def read_file_sync(f: File) -> str:
    with f.open_sync("rb") as fh:
        contents = fh.read().decode("utf-8")
        return contents

open_sync obtains the fsspec filesystem for self.path, passes the mode and compression, adds block_size when supplied, and applies cache_type and cache_options unless cache_type == "none". Match the API to the task's execution style: asynchronous code should use async with f.open(...), while synchronous code should use with f.open_sync(...).

Upload a local file or reference an existing one

When a local file already exists, use the matching upload factory:

@env.task
async def upload_local_file(content: str) -> File:
    with open("/tmp/data.csv", "w") as f:
        f.write("col1,col2\n1,2\n3,4\n")
    return await File.from_local("/tmp/data.csv")

@env.task

def upload_local_file_sync(content: str) -> File:
    with open("/tmp/data.csv", "w") as f:
        f.write("col1,col2\n1,2\n3,4\n")
    return File.from_local_sync("/tmp/data.csv")

Both methods first check that the source exists and raise ValueError("File not found: ...") otherwise. An explicit remote_destination selects the destination, for example:

@env.task
async def upload_to_specific_path() -> File:
    return await File.from_local(
        "/tmp/data.csv", "s3://my-bucket/data.csv"
    )

For a remote destination, the asynchronous implementation uses storage.put or storage.put_stream; the synchronous implementation writes through the underlying fsspec filesystem. A destination using the local file protocol is copied locally when an explicit destination is supplied. When no destination is supplied and the configured raw-data path is local, flyte-sdk can return the absolute local source path without copying.

If an object is already managed elsewhere, use a reference instead:

@env.task
async def process_existing_file(remote_path: str) -> str:
    f = File.from_existing_remote(remote_path)
    async with f.open("rb") as fh:
        content = bytes(await fh.read())
    return content.decode("utf-8")

from_existing_remote stores the URI and optional file_cache_key as hash; I/O occurs only when you call open, download, or exists. This is also the form used when a task output path is later injected into an application parameter.

Create deterministic remote artifacts

File.new_remote(file_name=None) asks internal_ctx().raw_data for a generated path. Without a filename, the generated path can have no extension. Supply a filename when its name matters:

@env.task
async def produce_file() -> File:
    f = File.new_remote(file_name="greeting.txt")
    async with f.open("wb") as fh:
        await fh.write(b"Hello from the produce_file task!")
    return f

For retry-safe checkpoints, use File.named_remote(name) instead. It requires an initialized Flyte runtime and an active task execution context. The method uses task_context.run_base_dir as its base. When the raw-data path ends in a value matching {run_name}-{node_id}-{attempt_index}, it extracts the node ID and includes it in the path. On a local filesystem it creates the parent directory and returns an absolute path; on a remote filesystem it constructs the path with that filesystem's separator. Pass a plain filename rather than a path containing separators.

The code-generation agent uses this behavior for checkpoint restoration:

for filename in _CHECKPOINT_FILES:
    remote = File.named_remote(f"{_task_hash}-{filename}")
    if await remote.exists():
        await remote.download(str(workspace / filename))

The same names are used when the checkpoint is written, so a later attempt can find the earlier artifact. Unlike named_remote, new_remote requests a generated path and is not the deterministic choice for this use case.

Download and check a file

Download methods return the actual local path. With no argument, flyte-sdk asks storage for a generated temporary local path:

@env.task
async def download_file(f: File) -> str:
    local_path = await f.download()
    with open(local_path, "r") as fh:
        return fh.read()

@env.task

def download_file_sync(f: File) -> str:
    local_path = f.download_sync()
    with open(local_path, "r") as fh:
        return fh.read()

An explicit non-directory path is treated as the exact output filename. An existing directory, or a path ending in the operating system separator, is treated as a directory and receives the source basename. Parent directories are created for local copies:

@env.task
async def download_to_path(f: File) -> str:
    return await f.download("/tmp/downloaded/my_custom_filename")

For local sources, download uses asynchronous local copying and download_sync uses shutil.copy2. For remote sources they call the asynchronous or synchronous storage get operation. Both normalize explicit paths to absolute paths while preserving a trailing separator, which is what lets directory targets retain their intended meaning.

Check existence without downloading the contents:

@env.task
async def check_file(f: File) -> bool:
    return await f.exists()

@env.task

def check_file_sync(f: File) -> bool:
    return f.exists_sync()

These delegate to flyte.storage.exists and flyte.storage.exists_sync, respectively.

Hashes, deferred uploads, and serialization

The optional hash_method argument supports two distinct cases. A string is treated as an already-computed cache key. A HashMethod object is used while bytes are written or transferred; flyte-sdk wraps transfers with HashingWriter or AsyncHashingReader and stores the resulting value in File.hash. PrecomputedValue is the special HashMethod case that supplies a value without reading data through a hashing wrapper.

A local upload can also be deferred. When there is no raw-data context and no explicit destination, from_local and from_local_sync return a File whose path is still local and whose private lazy_uploader is set. FileTransformer.to_literal invokes that uploader later. In local run mode it resolves to the local path; in remote mode it calls flyte.remote.upload_file.aio.

FileTransformer is registered with TypeEngine when flyte.io._file is imported. It does not generally read or upload file contents: it converts a File reference into a Flyte single-part blob containing the URI, format, and optional literal hash. Its reverse conversion requires a blob with single dimensionality, derives name from the URI, and reconstructs the expected File type. Non-blob and multipart literals are rejected. The transformer's get_literal_type currently emits an empty blob format; the source contains a TODO to derive that format from the generic type, even though the runtime blob metadata uses File.format.

The nested File.Config enables arbitrary_types_allowed and adds the schema metadata x-flyte-type: file, with the description “A file reference with an optional format type.”

Configure the storage context

Generated paths and automatic uploads follow the active Flyte raw-data configuration. Initialize normally with flyte.init(); tests that exercise an S3 backend use flyte.init(storage=S3.for_sandbox()). The same abstraction can use other fsspec-compatible backends. The Hugging Face example runs with an explicit raw-data path:

flyte.init_from_config()
run = flyte.with_runcontext(
    raw_data_path=f"hf://buckets/{USERNAME}/{BUCKET}/raw-data/"
).run(pipeline, name="HuggingFace")

For authenticated Hugging Face Storage Buckets, the example notes that the fsspec integration reads HF_TOKEN; remote execution must provide that environment variable, such as through a Flyte secret. The backend selected by initialization determines how open, upload, download, and existence operations resolve the File.path.

Troubleshoot the common edge cases

• Factory call fails before a task runs: initialize Flyte before calling context-sensitive factories. named_remote additionally raises ValueError when no active task execution context exists.
• The generated name has no suffix: new_remote() without file_name uses a generated path and tests allow an extensionless result. Pass file_name="data.csv" when needed.
• Decoded async data has the wrong type: convert the result of async read() with bytes(...) before calling .decode(...).
• A download appears under an unexpected name: an existing directory or separator-terminated target gets the source basename; use a non-directory path for an exact filename.
• An async remote open raises OnlyAsyncIOSupportedError: flyte-sdk falls back only for local filesystems. A non-local unsupported protocol re-raises the error.
• Calling from_local did not upload immediately: with no raw-data context and no explicit destination, upload is intentionally represented by a lazy uploader and is triggered during literal conversion.
• A constructed File did not copy anything: File(path=...) is a reference constructor. Use from_local for an upload or new_remote plus open for direct output.