Reading and Writing File Data

When passing files between tasks or interacting with blob storage in flyte-sdk, the system does not implicitly transfer or download file payloads in the background. The File class represents a reference to a local or remote file, and tasks explicitly control when and how bytes are streamed, downloaded, or uploaded.

The FileTransformer class handles metadata conversion into Flyte literals (as BlobType with SINGLE dimensionality) without performing any storage I/O during workflow transitions. All actual data transfer is managed directly through File methods.

import pandas as pd
from pandas import DataFrame
from flyte.io import File

# Async streaming directly to remote blob storage
async def write_remote_csv(df: pd.DataFrame) -> File[DataFrame]:
    remote_file = File[DataFrame].new_remote()
    async with remote_file.open("wb") as f:
        df.to_csv(f)
    return remote_file

# Async streaming read from a remote File reference
async def process_remote_csv(csv_file: File[DataFrame]) -> int:
    async with csv_file.open("rb") as f:
        df = pd.read_csv(f)
    return len(df)

---

Reading File Data

The File class supports streaming data over the network via fsspec and aiofiles, reading synchronously or asynchronously, or downloading the full file to local disk.

Asynchronous Streaming with open()

Use file.open() inside an async context manager to stream file contents without saving the entire file to disk first. For remote files, mode must include "b" (binary mode).

from flyte.io import File

async def read_file_chunks(remote_file: File) -> bytes:
    # mode defaults to 'rb'
    async with remote_file.open(mode="rb", block_size=64 * 1024, cache_type="readahead") as f:
        data = await f.read()
    return data

File.open() accepts parameters passed directly to the underlying fsspec or aiofiles layer:

• mode: File access mode (default "rb"). Must include "b" for remote storage backends.
• block_size: Size of blocks in bytes for buffered reading.
• cache_type: Caching strategy such as "readahead", "mmap", "bytes", or "none" (default "readahead").
• cache_options: Dictionary of options configuring the cache backend.
• compression: Explicit compression format (e.g., "gzip", "bz2") or None for auto-detection.

If the underlying fsspec filesystem does not implement open_async, File.open() catches NotImplementedError and falls back to running the file handle synchronously.

Synchronous Streaming with open_sync()

If a task or downstream library does not use asyncio, use the synchronous context manager file.open_sync():

from flyte.io import File

def read_sync(remote_file: File) -> str:
    with remote_file.open_sync("rb") as f:
        content = f.read()
    return content.decode("utf-8")

Downloading to Local Disk with download()

When using libraries or command-line utilities that require a local POSIX filesystem path, download the file using await file.download():

from pathlib import Path
from flyte.io import File

async def process_on_disk(remote_file: File) -> str:
    # Downloads to a temporary local path if local_path is omitted
    temp_local_path: str = await remote_file.download()

    # Or specify a target destination
    explicit_path: str = await remote_file.download(local_path="/tmp/input_data.bin")

    assert Path(explicit_path).exists()
    return explicit_path

If the File already references a local path (file:// protocol), download() copies the file locally via aiofiles rather than making a network request.

Checking Existence with exists_sync()

Verify whether a target file exists on local or remote storage before attempting to open it:

from flyte.io import File

def check_file(file: File) -> bool:
    return file.exists_sync()

---

Creating and Returning File Outputs

Tasks create and return File instances in three distinct ways depending on where the data originates and how it is written.

1. Direct Remote Streaming via new_remote()

File.new_remote() allocates a destination URI under the configured raw data prefix (ctx.raw_data.get_random_remote_path()). This avoids writing temporary files to local disk and uploading them afterward.

import pandas as pd
from pandas import DataFrame
from flyte.io import File

async def generate_dataset() -> File[DataFrame]:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})

    # Allocates a remote path in the configured storage prefix
    output_file = File[DataFrame].new_remote()

    # Stream bytes directly to the remote object store
    async with output_file.open("wb") as f:
        df.to_parquet(f)

    return output_file

2. Uploading Local Files via from_local()

When a task writes output to the local container disk or an external tool generates a local artifact, use await File.from_local() to upload it:

from pathlib import Path
from flyte.io import File

async def export_local_artifact() -> File:
    local_path = Path("/tmp/model_weights.bin")
    local_path.write_bytes(b"model-weights-data")

    # Uploads to an auto-generated path in raw data storage
    uploaded_file = await File.from_local(local_path)

    # Or specify an explicit destination URI
    custom_dest_file = await File.from_local(
        local_path=local_path,
        remote_destination="s3://my-storage-bucket/models/v1.bin"
    )

    return uploaded_file

If remote_destination is omitted and the storage backend protocol is local (file://), File.from_local() optimizes by referencing the existing absolute local path directly instead of creating a redundant file copy.

3. Referencing Existing Remote Files via from_existing_remote()

To return or forward a reference to an existing object without performing any read or write I/O:

from flyte.io import File

def point_to_raw_dataset() -> File:
    return File.from_existing_remote("s3://my-bucket/datasets/raw_data.parquet")

---

Task Caching and File Hashing

When tasks use caching, Flyte determines cache keys using the hash attribute on File objects.

File integrates with flyte.io._hashing_io accumulator mechanisms (HashMethod and PrecomputedValue):

1. Hashing during streaming: When writing to a File created with File.new_remote(hash_method=...), open() wraps the file handle in a HashingWriter. When the context manager exits, the accumulated digest is automatically assigned to file.hash.
2. Precomputed hashes: When calling File.from_local() or File.from_existing_remote(), pass a string hash or PrecomputedValue to supply the cache key directly.

from flyte.io import File
from flyte.io._hashing_io import PrecomputedValue

# Setting an explicit hash key on an existing remote reference
static_file = File.from_existing_remote(
    "s3://my-bucket/data.csv",
    file_cache_key="md5:c4ca4238a0b923820dcc509a6f75849b"
)

# Uploading a local file with a precomputed hash
async def upload_with_hash(local_path: str) -> File:
    return await File.from_local(
        local_path,
        hash_method=PrecomputedValue("sha256-digest-value")
    )

---

Troubleshooting and Gotchas

Remote Files Require Binary Mode in open()

Attempting to open a remote file with a text mode (e.g., mode="r" or mode="w") raises a ValueError:

ValueError: Mode must include 'b' for binary access, when using remote files.

Always use binary modes ("rb", "wb", "ab") when calling File.open() on remote paths. Decode text explicitly or pass the binary handle to readers that support binary streams (such as pandas.read_csv).

FileTransformer Does Not Perform Automatic I/O

In flyte-sdk, passing a File into a task does not download the file to the local container, and returning a File does not automatically upload local files unless File.from_local() or File.new_remote() was used. If a task writes to /tmp/data.csv and returns File(path="/tmp/data.csv") without from_local(), downstream tasks executing in separate pods will receive a local path reference that does not exist in their environment.

download() Is Async Only

File.download() is an asynchronous method (async def download(...)). There is no synchronous download_sync() method on File. If running in synchronous execution contexts, use open_sync("rb") to stream bytes into a local file:

from flyte.io import File

def download_sync(file: File, destination_path: str) -> None:
    with file.open_sync("rb") as src, open(destination_path, "wb") as dst:
        dst.write(src.read())

Context Initialization Requirement

Both File.new_remote() and File.from_local() use the @requires_initialization decorator. They access flyte._context.internal_ctx().raw_data to generate target paths. Ensure the Flyte environment is initialized before invoking these methods.