Remote Execution and Inspection

The flyte.remote package in flyte-sdk provides a robust interface for interacting with the Flyte API, enabling you to launch, inspect, and manage various Flyte entities remotely. This includes tasks, runs, projects, triggers, and applications, allowing for programmatic control over your Flyte workflows.

Launching and Monitoring Remote Runs

When you execute a task on Flyte, it results in a "Run." The flyte.remote.Run class allows you to interact with these remote executions, monitoring their progress, retrieving logs, and accessing their inputs and outputs.

Retrieving Existing Runs

To inspect a run that has already been launched, you can retrieve it by its name or list all available runs.

If you know the exact name of a run, use Run.get():

from flyte.remote import Run

# Assuming 'my-run-name' is the name of an existing run
# flyte.init_from_config() must be called first to configure the client
# (e.g., from a config file or environment variables)
# For this example, we assume flyte.init_from_config() has been called.

my_run = await Run.get("my-run-name")
print(f"Retrieved run: {my_run.name}, current phase: {my_run.phase}")

To discover runs, you can use Run.listall(), optionally filtering by phase, task name, or creation time. This method returns an asynchronous iterator of Run objects.

from flyte.remote import Run, TimeFilter
from flyte.remote._action import ActionPhase
from datetime import datetime, timedelta

# List all runs that are currently running or have succeeded
async for run in Run.listall(in_phase=(ActionPhase.RUNNING, ActionPhase.SUCCEEDED)):
    print(f"Run: {run.name}, Phase: {run.phase}, Task: {run.action.task_name}")

# List runs for a specific task that were created in the last 24 hours
yesterday = datetime.utcnow() - timedelta(days=1)
time_filter = TimeFilter()
time_filter.after = yesterday
async for run in Run.listall(task_name="my_specific_task", created_at=time_filter):
    print(f"Specific Task Run: {run.name}, Phase: {run.phase}")

Waiting for Run Completion

After launching a run (or retrieving an existing one), you often need to wait for it to reach a terminal state (success or failure). The Run.wait() method provides a convenient way to do this, offering visual feedback in interactive environments.

from flyte.remote import Run

# Assuming 'my_run' is an active Run object
await my_run.wait()
print(f"Run \'{my_run.name}\' finished with phase: {my_run.phase}")

# You can also wait for a specific state, like 'running'
# This will return as soon as the run enters the 'running' phase
# Note: This example assumes a new run or a run that has not yet reached 'running'
# For an already completed run, wait() will return immediately.
# my_run = await Run.get("another-run-name") # Re-fetch if needed
# await my_run.wait(wait_for="running")
# print(f"Run \'{my_run.name}\' is now running.")

Internally, Run.wait() delegates to self.action.wait(). The Action class, which represents the core execution unit within a Run, handles the polling and status updates. It uses ActionDetails.watch() to stream updates from the Flyte API, updating its internal pb2 (protobuf) representation and _details cache.

Streaming Run Updates

For more fine-grained control or to build custom monitoring UIs, you can use Run.watch() to receive real-time updates about a run's status. This method yields ActionDetails objects as the run progresses.

from flyte.remote import Run

# Assuming 'my_run' is an active Run object
print(f"Watching run \'{my_run.name}\'...")
async for details in my_run.watch():
    print(f"  Current phase: {details.phase}, Attempts: {details.attempts}, Runtime: {details.runtime}")
    if details.done():
        print(f"Run \'{my_run.name}\' is complete.")
        break

The Run.watch() method internally calls self.action.watch(), which in turn uses ActionDetails.watch() to establish a streaming connection to the Flyte API. This allows the Run object's internal state, including run.action.pb2, to be continuously updated.

Accessing Run Logs

To view the logs generated by a remote run, use Run.show_logs() for an interactive display or Run.get_logs() to retrieve logs programmatically as an asynchronous generator of strings.

from flyte.remote import Run

# Assuming 'my_run' is an active Run object
print(f"Displaying logs for run \'{my_run.name}\'")
# await my_run.show_logs(max_lines=50, show_ts=True) # Uncomment to display logs interactively

print(f"Retrieving logs programmatically for run \'{my_run.name}\'")
async for line in my_run.get_logs(filter_system=True):
    print(f"  {line}")

Both show_logs() and get_logs() delegate to the corresponding methods on the Run's underlying Action object. The Action.get_logs() method uses flyte.remote._logs._format_line to format log entries, optionally adding timestamps or filtering system messages.

Terminating a Run

If you need to stop a running task, you can abort the run using Run.abort():

from flyte.remote import Run

# Assuming 'my_run' is an active Run object
# await my_run.abort(reason="User requested early termination.")
# print(f"Run \'{my_run.name}\' has been aborted.")

This method sends an AbortRunRequest to the Flyte API via get_client().run_service.abort_run, effectively terminating the remote execution.

Accessing Run Inputs and Outputs

Once a run is complete, you can retrieve the inputs it received and the outputs it produced. These are available via Run.inputs() and Run.outputs(), respectively.

from flyte.remote import Run

# Assuming 'my_run' is a completed Run object
# Ensure the run is complete before attempting to get outputs
await my_run.wait()

run_inputs = await my_run.inputs()
print(f"Run inputs: {run_inputs}")

run_outputs = await my_run.outputs()
print(f"Run outputs: {run_outputs}")

# Accessing named outputs (if available)
# print(f"Named outputs: {run_outputs.named_outputs}")

# Accessing outputs by index (tuple-like behavior)
# if len(run_outputs) > 0:
#     print(f"First output: {run_outputs[0]}")

Both Run.inputs() and Run.outputs() internally call Run.details().inputs() and Run.details().outputs(). The RunDetails object then delegates to ActionDetails.inputs() and ActionDetails.outputs(). These methods fetch the data from the dataproxy_service and convert the protobuf representations into native Python types using flyte._internal.runtime.convert.

ActionInputs extends UserDict, allowing dictionary-like access to inputs. ActionOutputs extends tuple, providing tuple-like access (indexing, unpacking) and also offers a named_outputs property for dictionary-style access when output names are available.

from flyte.remote import ActionOutputs
from flyteidl2.common import common_pb2

# Example of ActionOutputs behavior (from tests/flyte/remote/test_action_io.py)
outputs_pb2 = common_pb2.Outputs() # Placeholder protobuf
test_data = (42, "hello", [1, 2, 3], {"flag": True}, None)
outputs = ActionOutputs(pb2=outputs_pb2, data=test_data)

assert len(outputs) == 5
assert outputs[0] == 42
assert outputs[1] == "hello"
assert outputs[2] == [1, 2, 3]
assert outputs[3] == {"flag": True}
assert outputs[4] is None

Managing Flyte Entities

The flyte.remote package provides classes for managing various Flyte entities, including tasks, projects, triggers, and applications.

Tasks

The flyte.remote.Task and flyte.remote.TaskDetails classes allow you to retrieve and inspect task definitions deployed on Flyte.

To get a task definition, use Task.get(). This method returns a LazyEntity, which defers fetching the full task details until explicitly requested with .fetch.aio(). This pattern is useful for avoiding unnecessary network calls.

from flyte.remote import Task
from flyte.errors import RemoteTaskNotFoundError

task_name = "my_remote_task"
project_name = "flyte-project"
domain_name = "development"

try:
    # Get the lazy entity - this doesn't fetch yet
    lazy_task = Task.get(task_name, project=project_name, domain=domain_name, auto_version="latest")

    # Explicitly fetch to retrieve full task details
    task_details = await lazy_task.fetch.aio()
    print(f"Fetched task: {task_details.name}, version: {task_details.version}, type: {task_details.task_type}")
    print(f"Required arguments: {task_details.required_args}")

except RemoteTaskNotFoundError as e:
    print(f"Error: {e}")

Task.get() supports version or auto_version parameters. auto_version="latest" fetches the most recently created version of the task, while auto_version="current" is used within a task context to refer to the version of the currently executing task.

To discover tasks, use Task.listall():

from flyte.remote import Task

# List all tasks in the current project/domain
async for task in Task.listall():
    print(f"  - Task: {task.name}, Version: {task.version}")

# List tasks by a specific name
async for task in Task.listall(by_task_name="my_specific_task"):
    print(f"  - Specific Task: {task.name}, Version: {task.version}")

TaskDetails provides properties to inspect various aspects of the task definition, such as interface (input/output types), cache policy, secrets required, and resources requested.

# Assuming task_details is a TaskDetails object
# print(f"Task interface: {task_details.interface}")
# print(f"Cache policy: {task_details.cache.behavior}")
# print(f"Secrets required: {task_details.secrets}")
# print(f"Resources: {task_details.resources}")

Overriding Task Properties

You can create a new TaskDetails instance with overridden properties using the override() method. This is useful for adjusting task behavior without modifying the original task definition.

from flyte.remote import TaskDetails, Task
import flyte

# Assuming lazy_task is a LazyEntity for a Task
task_details = await lazy_task.fetch.aio()

# Override the task's short name and add retries
overridden_task = task_details.override(
    short_name="new_short_name",
    retries=3,
)
print(f"Original task short name: {task_details.pb2.spec.short_name}")
print(f"Overridden task short name: {overridden_task.pb2.spec.short_name}")
print(f"Overridden task retries: {overridden_task.pb2.spec.task_template.metadata.retries.retries}")

Gotcha: Remote tasks cannot be executed locally. Attempting to call a TaskDetails object directly outside of a Flyte task context will raise a flyte.errors.RemoteTaskUsageError.

# This will raise RemoteTaskUsageError if not in a task context
# await task_details(input_arg="value")

Projects

The flyte.remote.Project class allows you to manage projects within Flyte, including creation, retrieval, and state changes (archiving/unarchiving).

To create a new project:

from flyte.remote import Project

# await Project.create(id="my-new-project-id", name="My New Project", description="A project for documentation.")
# print("Project created successfully.")

To retrieve a project:

from flyte.remote import Project

# my_project = await Project.get("my-new-project-id")
# print(f"Retrieved project: {my_project.pb2.name}, State: {my_project.pb2.state}")

You can update project details or change its state:

from flyte.remote import Project

# Assuming 'my_project' is an existing Project object
# updated_project = await Project.update(id=my_project.pb2.id, description="Updated description.")
# print(f"Project description updated to: {updated_project.pb2.description}")

# Archive a project
# archived_project = my_project.archive()
# print(f"Project \'{archived_project.pb2.name}\' archived.")

# Unarchive a project
# unarchived_project = my_project.unarchive()
# print(f"Project \'{unarchived_project.pb2.name}\' unarchived.")

To list all projects, optionally filtering by their archived status:

from flyte.remote import Project

print("Active projects:")
async for project in Project.listall():
    print(f"  - {project.pb2.name} (ID: {project.pb2.id})")

print("Archived projects:")
async for project in Project.listall(archived=True):
    print(f"  - {project.pb2.name} (ID: {project.pb2.id})")

Triggers

The flyte.remote.Trigger class enables you to create, manage, and inspect scheduled executions or event-driven triggers for your tasks.

To create a new trigger, you first define a flyte.Trigger object and then use Trigger.create():

import flyte
from flyte.remote import Trigger

# Assuming 'my_task_name' exists and flyte.init_from_config() has been called
# trigger_obj = flyte.Trigger(
#     name="my-daily-trigger",
#     automation=flyte.Cron(schedule="0 0 * * *"), # Daily at midnight UTC
#     description="Triggers my_task_name daily",
#     auto_activate=True,
#     inputs={"trigger_time": flyte.TriggerTime} # Example of passing trigger time as input
# )
# new_trigger = await Trigger.create(trigger_obj, task_name="my_task_name")
# print(f"Trigger \'{new_trigger.name}\' created for task \'{new_trigger.task_name}\'")

To retrieve a specific trigger:

from flyte.remote import Trigger

# my_trigger_details = await Trigger.get(name="my-daily-trigger", task_name="my_task_name")
# print(f"Retrieved trigger: {my_trigger_details.name}, Active: {my_trigger_details.is_active}")

To list all triggers, optionally filtered by task:

from flyte.remote import Trigger

print("All triggers:")
async for trigger in Trigger.listall():
    print(f"  - {trigger.name} (Task: {trigger.task_name}, Active: {trigger.is_active})")

print("Triggers for 'my_task_name':")
async for trigger in Trigger.listall(task_name="my_task_name"):
    print(f"  - {trigger.name} (Active: {trigger.is_active})")

You can update a trigger's active state or delete it:

from flyte.remote import Trigger

# await Trigger.update(name="my-daily-trigger", task_name="my_task_name", active=False)
# print("Trigger deactivated.")

# await Trigger.delete(name="my-daily-trigger", task_name="my_task_name")
# print("Trigger deleted.")

Applications

The flyte.remote.App class allows you to deploy and manage applications on Flyte, controlling their lifecycle and inspecting their status.

To retrieve an application:

from flyte.remote import App

# my_app = await App.get(name="my-flyte-app")
# print(f"App: {my_app.name}, Endpoint: {my_app.endpoint}, Status: {my_app.deployment_status}")

You can activate (start) or deactivate (stop) an application:

from flyte.remote import App

# my_app = await App.get(name="my-flyte-app")

# if not my_app.is_active():
#     activated_app = await my_app.activate(wait=True)
#     print(f"App \'{activated_app.name}\' activated.")

# if not my_app.is_deactivated():
#     deactivated_app = await my_app.deactivate(wait=True)
#     print(f"App \'{deactivated_app.name}\' deactivated.")

To monitor an application's deployment status, use App.watch():

from flyte.remote import App

# my_app = await App.get(name="my-flyte-app")
# print(f"Watching app \'{my_app.name}\'...")
# watched_app = await my_app.watch(wait_for="activated")
# print(f"App \'{watched_app.name}\' is now activated.")

To update an application's specification, use App.replace():

from flyte.remote import App
from flyteidl2.app import app_definition_pb2

# Assuming an existing app 'test-app' and a new spec `updated_spec`
# updated_spec = app_definition_pb2.AppSpec( # ... populate with new spec details ... )
# result_app = await App.replace.aio(
#     name="test-app",
#     updated_app_spec=updated_spec,
#     reason="Updating app configuration",
# )
# print(f"App \'{result_app.name}\' updated to revision {result_app.revision}.")

Gotcha: When using App.replace(), if the updated_app_spec does not explicitly define ingress configuration, the existing ingress from the old application specification will be used for comparison. An AssertionError will be raised if the original App object's spec itself lacks ingress information during this comparison.

Data Upload for Remote Tasks

When your remote tasks require local files or directories, the flyte.remote package provides utility functions to upload them to remote storage.

Uploading a Single File

Use flyte.remote.upload_file() to upload a single file. It returns the MD5 digest and the remote native URL of the uploaded file.

import flyte.remote
from pathlib import Path

# Create a dummy file for demonstration
dummy_file_path = Path("my_local_file.txt")
dummy_file_path.write_text("Hello, Flyte remote storage!")

# hash_digest, remote_url = await flyte.remote.upload_file.aio(dummy_file_path)
# print(f"Uploaded file: {dummy_file_path} to {remote_url} (MD5: {hash_digest})")

# Clean up the dummy file
dummy_file_path.unlink()

Uploading a Directory

Use flyte.remote.upload_dir() to upload an entire local directory. It returns the remote URI of the uploaded directory.

import flyte.remote
from pathlib import Path

# Create a dummy directory with some files
dummy_dir_path = Path("my_local_dir")
dummy_dir_path.mkdir(exist_ok=True)
(dummy_dir_path / "file1.txt").write_text("Content of file 1")
(dummy_dir_path / "subdir").mkdir(exist_ok=True)
(dummy_dir_path / "subdir" / "file2.txt").write_text("Content of file 2")

# remote_dir_uri = await flyte.remote.upload_dir.aio(dummy_dir_path)
# print(f"Uploaded directory: {dummy_dir_path} to {remote_dir_uri}")

# Clean up the dummy directory
import shutil
shutil.rmtree(dummy_dir_path)

Both upload_file() and upload_dir() utilize an internal _upload_with_retry mechanism. This mechanism automatically retries uploads in case of transient network issues like timeouts, connection errors, or server-side 5xx errors. However, it does not retry on client-side errors (e.g., 4xx status codes).

Configuration for Uploads

The default read timeout for file uploads is 600 seconds (10 minutes). You can override this value by setting the FLYTE_UPLOAD_TIMEOUT environment variable:

export FLYTE_UPLOAD_TIMEOUT=1200 # Set timeout to 20 minutes

This is particularly useful in environments with slower network connections or for uploading very large files.