Skip to main content

Advanced Input and Output Handling

Pass typed data through a custom container​

To run a raw container while preserving Flyte input and output types, construct flyte.extras.ContainerTask with typed inputs and outputs, and make the container use the configured input and output mount points:

import datetime

from flyte.extras import ContainerTask
from flyte.io import Dir, File

container_task = ContainerTask(
    name="convert-data",
    image="python:3.11",
    command=[
        "python",
        "convert.py",
        "/var/inputs/source",
        "/var/inputs/config_dir",
        "--limit",
        "{{.inputs.limit}}",
    ],
    inputs={
        "source": File,
        "config_dir": Dir,
        "limit": int,
    },
    outputs={
        "result": File,
        "rows": int,
        "finished": datetime.datetime,
    },
)

This is a usage pattern derived from ContainerTask in extras/_container.py; the checkout does not contain a repository-local ContainerTask(...) call site. flyte.extras re-exports the class from extras/__init__.py, so from flyte.extras import ContainerTask is the public import.

The constructor creates a NativeInterface from inputs and outputs, sets the task type to raw-container, and converts string input and output directories to pathlib.Path objects. The defaults are /var/inputs and /var/outputs. A string image is also converted: "auto" uses Image.from_debian_base(), and any other string uses Image.from_base(image).

Substitute scalar inputs in command arguments​

For ordinary inputs, use the template form {{.inputs.name}} in a command or argument. _render_command_and_volume_binding obtains the matching keyword argument and replaces the template with str(input_val):

from flyte.extras import ContainerTask

scalar_task = ContainerTask(
    name="write-message",
    image="alpine:3.18",
    command=["sh", "-c", "echo {{.inputs.message}} > /var/outputs/message"],
    inputs={"message": str},
    outputs={"message": str},
)

_prepare_command_and_volumes applies this rendering independently to every item in the concatenation of command and arguments. Missing keyword values are not explicitly rejected by the renderer; a missing value used by a template is converted to the string "None".

Mount File and Dir inputs​

File and Dir inputs use a different syntax. Put the input name in a path beneath input_data_dir, such as /var/inputs/source. The renderer extracts the key from that path and creates a Docker binding from the local input_val.path to the corresponding container path:

from flyte.extras import ContainerTask
from flyte.io import Dir, File

file_task = ContainerTask(
    name="inspect-inputs",
    image="alpine:3.18",
    command=[
        "sh",
        "-c",
        "test -f /var/inputs/source && test -d /var/inputs/config_dir && "
        "cat /var/inputs/source > /var/outputs/contents",
    ],
    inputs={"source": File, "config_dir": Dir},
    outputs={"contents": str},
)

For these exact File and Dir values, the resulting binding has this shape:

{
    local_flyte_file_or_dir_path: {
        "bind": "/var/inputs/source",
        "mode": "rw",
    }
}

The actual key used for the second input is /var/inputs/config_dir. Both input and output Docker mounts use "rw" mode. The implementation checks type(input_val) in [File, Dir], not isinstance, so subclasses or specialized wrappers do not take this volume-mount branch.

Do not write a File or Dir reference as {{.inputs.source}}. The implementation raises an AssertionError and instructs callers to use a path-like reference under the configured input directory. The path extractor recognizes one simple name after the input directory; names are limited to word characters, hyphens, and periods by the extractor, so nested paths and more elaborate shell expressions are not supported by this mechanism.

To use another mount location, configure both the command path and input_data_dir consistently:

from flyte.extras import ContainerTask
from flyte.io import File

custom_mount_task = ContainerTask(
    name="custom-input-root",
    image="alpine:3.18",
    command=["cat", "/work/infile", ">", "/work/outfile"],
    inputs={"infile": File},
    outputs={"outfile": File},
    input_data_dir="/work",
    output_data_dir="/work",
)

input_data_dir is also the prefix used to recognize the File/Dir path and the container-side destination is formed with os.path.join(input_data_dir, key).

Write and collect outputs​

During execute, ContainerTask creates a random host directory with storage.get_random_local_directory() and mounts it at output_data_dir. A declared output named result must therefore be written by the container at the host-side output directory’s result entry, which appears inside the container as <output_data_dir>/result:

from flyte.extras import ContainerTask

output_task = ContainerTask(
    name="make-output",
    image="alpine:3.18",
    command=["sh", "-c", "printf '42' > /var/outputs/number"],
    outputs={"number": int},
)

_get_output iterates through the declared output dictionary in order. For each key it looks at <output_directory>/<key>. Regular files are read completely as text; a non-file path produces None before type conversion. The returned value is a tuple, deliberately allowing Flyte to map each dictionary-declared output to a separate task result.

Output conversion is based on the declared Python type, not on metadata_format:

  • bool returns False only when the text, lowercased, is exactly "false"; every other text is True.
  • datetime.datetime uses datetime.datetime.fromisoformat.
  • datetime.timedelta uses _string_to_timedelta, which accepts forms such as HH:MM:SS, MM:SS, and days, HH:MM:SS[.microseconds].
  • File calls await File.from_local(output_path).
  • Dir calls await Dir.from_local(output_path).
  • Other types are constructed by calling the type with the output text, as with int(output_val) or str(output_val).

File.from_local validates that the local path exists and integrates the result with Flyte’s configured storage. Consequently, a container producing a File output must create the output path named by that output key. A missing boolean output, for example, reaches output_val.lower() and fails because the missing value is None; other output types can likewise fail when constructed from None.

Configure serialization and raw-container data loading​

Use metadata_format to select the raw-container data-loading format sent to Flyte:

from flyte.extras import ContainerTask

yaml_task = ContainerTask(
    name="yaml-container",
    image="example/image:latest",
    command=["/app/run"],
    metadata_format="YAML",
    input_data_dir="/var/inputs",
    output_data_dir="/var/outputs",
)

MetadataFormat is Literal["JSON", "YAML", "PROTO"]. data_loading_config always sets enabled=True, includes the normalized input and output paths, and maps those three strings to the corresponding tasks_pb2.DataLoadingConfig enum values. The implementation’s mapping uses JSON as the fallback for an unrecognized value.

This setting does not make _get_output parse JSON, YAML, or protobuf content. _get_output reads output files as text and converts them according to the declared output type. metadata_format configures the serialized raw-container data-loading message instead.

When the runtime serializes the task, _internal/runtime/task_serde.py calls the two ContainerTask hooks in _get_urun_container:

return tasks_pb2.Container(
    image=img_uri,
    command=[],
    args=task_template.container_args(serialize_context),
    resources=resources,
    env=env,
    data_config=task_template.data_loading_config(serialize_context),
    config=task_template.config(serialize_context),
)

container_args returns self._cmd + (self._args if self._args else []), so arguments are appended to command for serialization as well as for local execution. The serializer resolves the image through task_template.image.uri; a string image at that stage is rejected, which is why the constructor’s conversion of string images to Image objects matters.

Understand local execution behavior​

execute is the local Docker path. It lazily imports the optional docker package, normalizes the two mount paths, prepares the command and input bindings, and adds the output binding:

volume_bindings[str(output_directory)] = {
    "bind": self._output_data_dir,
    "mode": "rw",
}

It then obtains a client with docker.from_env(), resolves the Image.uri, checks the local image list, and pulls the image when it is absent. The container is started detached with remove=True and the complete volume map. The rendered command is printed unconditionally. Container logs are streamed to stdout only when local_logs=True; the default is True. After container.wait(), output collection begins.

For local execution, install the optional Docker dependency and provide a reachable Docker daemon. If the Python package is missing, execute raises:

Docker is not installed. Please install Docker by running `pip install docker`.

Operational limitations​

  • container.wait() has no timeout; the source contains a TODO for adding one and the wait can block indefinitely.
  • The returned container status is not inspected. A container failure can therefore be followed by output collection rather than an immediate status-based error.
  • The output mount and File/Dir input mounts are read-write ("rw").
  • The rendered command is printed even when local_logs=False, so command strings and substituted values can appear in local output.
  • File and directory inputs must be exact File or Dir instances for volume binding, and must use a path under input_data_dir rather than template syntax.
  • Declared outputs must be named files or directories under the output directory. _get_output does not parse serialized JSON, YAML, or protobuf files itself.
  • execute asserts that the resolved image is an Image object before reading .uri; despite the constructor accepting str | Image, execution depends on string conversion having occurred and on the resulting image being resolvable.