Fetching and Inspecting Remote Tasks

Fetching Remote Tasks

When you need to interact with a task that has already been registered on the Flyte platform, the flyte-sdk provides a mechanism to fetch its definition. This is particularly useful when you want to execute a pre-existing task or inspect its properties without redefining it in your local code.

To fetch a remote task, use the get class method of the remote.Task class. This method requires the task's name and either its version or an auto_version strategy. The auto_version parameter allows you to specify "latest" to retrieve the most recently created version of the task, or "current" to derive the version from the current task context, which is useful in deployment scenarios where all environments share the same version.

from flyte.remote import Task

# Fetch a task by its explicit version
my_task_v1 = Task.get(name="my_remote_task", version="v1.0.0", project="flyte-project", domain="development")

# Fetch the latest version of a task
my_task_latest = Task.get(name="my_remote_task", auto_version="latest", project="flyte-project", domain="development")

The Task.get() method, defined in remote/_task.py, does not immediately return the full task definition. Instead, it returns an instance of remote._task.LazyEntity.

Understanding LazyEntity: Deferred Loading

The LazyEntity class, found in remote/_task.py, acts as a proxy for the actual task details. This design defers the network call to the Flyte backend until the task's detailed information is actually needed. This can improve performance by avoiding unnecessary data fetches.

An instance of LazyEntity holds the task's name and a getter function that knows how to retrieve the TaskDetails. The actual TaskDetails object is fetched only when one of the following occurs:

1. You explicitly call the fetch() method on the LazyEntity.
2. You attempt to invoke the LazyEntity (e.g., by calling it like a function).
3. You access an attribute that requires the underlying TaskDetails (though direct attribute access on LazyEntity is not the primary mechanism for inspection).

To explicitly retrieve the TaskDetails object from a LazyEntity:

from flyte.remote import Task

# my_task_v1 is a LazyEntity instance
my_task_v1_lazy = Task.get(name="my_remote_task", version="v1.0.0", project="flyte-project", domain="development")

# Explicitly fetch the TaskDetails
task_details = my_task_v1_lazy.fetch()
print(f"Fetched task details for: {task_details.name}")

Internally, the LazyEntity.fetch() method uses a mutex to ensure that the task details are fetched only once, even if fetch() is called multiple times concurrently. If the _task attribute is None, it calls the stored _getter function to populate it.

Inspecting TaskDetails: Accessing Task Metadata

Once you have a TaskDetails object, you can inspect various properties of the remote task. The TaskDetails class, defined in remote/_task.py, exposes a rich set of attributes that provide insights into the task's configuration, interface, and resource requirements.

Here are some key properties you can access:

• name: The name of the task.
• version: The version of the task.
• task_type: The underlying type of the task (e.g., "python-task").
• interface: A NativeInterface object describing the task's inputs and outputs, including their types. This property is a functools.cached_property, meaning it's computed once and then cached for subsequent access.
  • required_args: A tuple of input argument names that do not have default values.
  • default_input_args: A tuple of input argument names that have default values.
• resources: A tuple containing the task's resource requests and limits (e.g., CPU, memory).
• cache: A flyte.Cache object detailing the task's caching policy, including behavior ("disable", "override", "auto"), version override, and ignored inputs.
• secrets: A list of secret keys configured for the task.

from flyte.remote import Task

my_task_lazy = Task.get(name="my_remote_task", version="v1.0.0", project="flyte-project", domain="development")
task_details = my_task_lazy.fetch()

print(f"Task Name: {task_details.name}")
print(f"Task Version: {task_details.version}")
print(f"Task Type: {task_details.task_type}")

print("--- Interface ---")
print(f"Required Arguments: {task_details.required_args}")
print(f"Default Input Arguments: {task_details.default_input_args}")
# The full interface object provides more details on types
# print(task_details.interface.inputs)

print("--- Resources ---")
requests, limits = task_details.resources
print(f"Resource Requests: {requests}")
print(f"Resource Limits: {limits}")

print("--- Caching ---")
print(f"Cache Behavior: {task_details.cache.behavior}")
print(f"Cache Version Override: {task_details.cache.version_override}")

print("--- Secrets ---")
print(f"Configured Secrets: {task_details.secrets}")

Invoking Remote Tasks

Both LazyEntity and TaskDetails objects are callable, allowing you to invoke the remote task as if it were a local Python function. When you call a LazyEntity instance, it first fetches the TaskDetails (if not already fetched) and then forwards the call to the underlying TaskDetails object's __call__ method.

from flyte.remote import Task

my_task_lazy = Task.get(name="my_remote_task", version="v1.0.0", project="flyte-project", domain="development")

# Invoke the remote task with keyword arguments
# This will implicitly fetch the TaskDetails if not already done
result = my_task_lazy(input_param_1="value1", input_param_2=123)
print(f"Remote task invocation result: {result}")

Important Considerations for Invocation

• Remote Execution Mode: Remote tasks (instances of LazyEntity or TaskDetails) can only be run when the flyte-sdk is operating in a 'remote' execution mode. Attempting to invoke a remote task outside of this mode will raise a ValueError.

  # Example of the error if not in remote mode
  # if isinstance(task, LazyEntity) and self._mode != "remote":
  #     raise ValueError("Remote task can only be run in remote mode.")

  This check is performed internally within the flyte._run module to ensure that remote tasks are executed in the appropriate context.

• Keyword Arguments Only: The __call__ method for remote tasks currently supports only keyword arguments. Positional arguments are not supported and will raise a flyte.errors.ReferenceTaskError.

  # This will raise an error:
  # result = my_task_lazy("value1", 123) 
  # flyte.errors.ReferenceTaskError: Reference task my_remote_task does not support positional arguments currently. Please use keyword arguments.

Overriding Task Parameters for Execution

Sometimes you may need to modify certain execution-specific parameters of a remote task without altering its registered definition on the Flyte platform. The override() method, available on both LazyEntity and TaskDetails, allows you to do this.

When you call override() on a LazyEntity, it first fetches the TaskDetails and then applies the overrides to that fetched instance. The override() method returns the modified LazyEntity (or TaskDetails) instance, allowing for chaining.

You can override parameters such as:

• short_name: A display name for the task execution.
• resources: Custom resource requests and limits for the execution (e.g., flyte.Resources(cpu="1", mem="500Mi")).
• retries: The number of retries for the task execution (can be an integer or a flyte.RetryStrategy).
• timeout: The maximum duration for the task execution (e.g., datetime.timedelta(minutes=5)).
• env_vars: Environment variables to set for the task's container.
• secrets: Specific secrets to be made available to the task.

from flyte.remote import Task
import datetime
import flyte

my_task_lazy = Task.get(name="my_remote_task", version="v1.0.0", project="flyte-project", domain="development")

# Override resources and add an environment variable
overridden_task = my_task_lazy.override(
    resources=flyte.Resources(cpu="2", mem="1Gi"),
    retries=3,
    timeout=datetime.timedelta(minutes=10),
    env_vars={"MY_ENV_VAR": "custom_value"}
)

# Now invoke the task with the overridden parameters
result = overridden_task(input_param="test_value")
print(f"Invoked task with overrides. Result: {result}")

# You can also inspect the overridden properties on the TaskDetails object
task_details_with_overrides = overridden_task.fetch()
print(f"Overridden CPU request: {task_details_with_overrides.resources[0].cpu}")

It's important to note that the override method only accepts specific keyword arguments. Passing unrecognized keyword arguments will result in a ValueError."