Discover and Fetch Remote Tasks

When working with remote Flyte tasks, flyte-sdk provides distinct mechanisms for discovering available tasks and retrieving their detailed configurations. This distinction optimizes performance by allowing you to work with lightweight references until comprehensive details are explicitly required.

Lightweight Task References with flyte.remote.Task

The flyte.remote.Task class represents a lightweight reference to a remote Flyte task. It contains only essential identifying information like its name and version, but not the full task definition, interface, or resource requirements. This makes Task objects efficient for listing and basic identification without incurring the overhead of fetching complete details from the Flyte backend.

Discovering Tasks with Task.listall

To find tasks within your Flyte project and domain, use the Task.listall class method. This method allows you to iterate through available tasks, optionally applying filters and sorting criteria. It returns an asynchronous iterator of Task objects.

For example, to list all tasks in the current project and domain, you can use:

from flyte.remote import Task

async def list_all_tasks():
    print("Listing all tasks:")
    async for task in Task.listall():
        print(f"- {task.name} (version: {task.version})")

# To run this in a synchronous context:
# import asyncio
# asyncio.run(list_all_tasks())

You can refine your search using parameters like by_task_name to filter by a specific task name, by_task_env for environment prefixes, project and domain to specify a different scope, and sort_by to order the results. The limit parameter controls the maximum number of tasks returned.

from flyte.remote import Task

async def list_specific_tasks(project_name: str, domain_name: str):
    print(f"Listing tasks named 'my_task' in project '{project_name}', domain '{domain_name}':")
    async for task in Task.listall(
        by_task_name="my_task",
        project=project_name,
        domain=domain_name,
        sort_by=("created_at", "desc"),
        limit=5,
    ):
        print(f"- {task.name} (version: {task.version}, created: {task.pb2.metadata.created_at})")

Internally, Task.listall constructs a ListTasksRequest protobuf message and sends it to the Flyte service's task_service.list_tasks endpoint. It handles pagination automatically, fetching tasks in batches up to the specified limit.

Referencing a Specific Task with Task.get (Lazy Entity)

When you know the name of a specific task you want to interact with, use Task.get. This method does not immediately fetch the full task details. Instead, it returns a LazyEntity object, which is a placeholder that will fetch the TaskDetails only when they are actually needed.

Task.get requires either an explicit version or an auto_version strategy:

• Explicit Version: Provide the exact version string if you know it.
• auto_version='latest': This instructs flyte-sdk to automatically resolve to the latest-by-time ordered version of the task in the specified project and domain. This is useful when you always want to use the most recently deployed version.
• auto_version='current': This option is designed for use within a running Flyte task context. It resolves the task version based on the version of the currently executing task. This ensures that a task always refers to other tasks within the same deployment version, promoting consistency in complex workflows. Note: Using auto_version='current' outside of a task context will raise a ValueError.

Here's how you might get a lazy reference to the latest version of a task:

from flyte.remote import Task

# Assuming 'my_project' and 'development' are configured or passed
task_name = "my_example_task"
project = "flyte-examples"
domain = "development"

# Get a lazy reference to the latest version of 'my_example_task'
lazy_task_ref = Task.get(task_name, project=project, domain=domain, auto_version="latest")

print(f"Created a lazy reference for task: {lazy_task_ref.name}")
# At this point, no remote call has been made to fetch full task details.

Comprehensive Task Details with flyte.remote.TaskDetails

The flyte.remote.TaskDetails class provides a complete representation of a remote task, including its full definition, input/output interfaces, resource requirements, caching strategy, and more. This object is necessary when you need to inspect the task's internal configuration or prepare it for execution.

Fetching Full Details from a LazyEntity

To retrieve the full TaskDetails from a LazyEntity obtained via Task.get, you must explicitly call its fetch() method (or fetch.aio() for asynchronous operations). This is when the actual remote call to the Flyte backend occurs.

If the task cannot be found (e.g., incorrect name, project, domain, or version), calling fetch() will raise a flyte.errors.RemoteTaskNotFoundError.

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

async def get_and_fetch_task(task_name: str, project: str, domain: str):
    try:
        lazy_task = Task.get(task_name, project=project, domain=domain, auto_version="latest")
        task_details = await lazy_task.fetch.aio()
        print(f"Successfully fetched details for task '{task_name}' (version: {task_details.version})")
        print(f"  - Type: {task_details.task_type}")
        print(f"  - Required arguments: {task_details.required_args}")
    except RemoteTaskNotFoundError:
        print(f"Error: Task '{task_name}' not found in project '{project}', domain '{domain}'.")

# Example usage (assuming 'my_example_task' exists):
# import asyncio
# asyncio.run(get_and_fetch_task("my_example_task", "flyte-examples", "development"))

As shown in src/flyte/remote/_task.py, the TaskDetails.get method (which Task.get delegates to) handles the logic for resolving auto_version and then constructs a LazyEntity with a deferred_get callable. This deferred_get function performs the actual get_task_details gRPC call to the Flyte service when fetch() is invoked.

Inspecting Task Metadata and Configuration

Once you have a TaskDetails object, you can access various properties to understand the task's configuration:

• name and version: The task's identifier.
• 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.
• required_args: A tuple of input argument names that do not have default values and must be provided during execution.
• default_input_args: A tuple of input argument names that have default values.
• cache: A flyte.Cache object detailing the task's caching behavior, including whether it's discoverable, its discovery version, and ignored inputs.
• secrets: A list of secret keys required by the task, as configured in its security_context.
• resources: A tuple containing the resource requests and limits (e.g., CPU, memory) for the task's container.

from flyte.remote import Task

async def inspect_task_details(task_name: str, project: str, domain: str):
    lazy_task = Task.get(task_name, project=project, domain=domain, auto_version="latest")
    task_details = await lazy_task.fetch.aio()

    print(f"Details for task '{task_details.name}' (version: {task_details.version}):")
    print(f"  Task Type: {task_details.task_type}")
    print(f"  Interface: {task_details.interface}")
    print(f"  Required Arguments: {task_details.required_args}")
    print(f"  Default Input Arguments: {task_details.default_input_args}")
    print(f"  Cache Policy: {task_details.cache.behavior} (version: {task_details.cache.version_override})")
    print(f"  Required Secrets: {task_details.secrets}")
    print(f"  Resources (Requests, Limits): {task_details.resources}")

# Example usage:
# import asyncio
# asyncio.run(inspect_task_details("my_example_task", "flyte-examples", "development"))

Dynamically Overriding Task Properties

Sometimes you need to execute a remote task with modified properties without altering its original definition on the Flyte backend. The TaskDetails.override() method allows you to create a new TaskDetails object with specific properties adjusted. This method is immutable; it does not modify the original TaskDetails instance.

You can override properties such as short_name, resources, retries, timeout, env_vars, secrets, max_inline_io_bytes, cache, and queue.

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte._secret.Secret' accepts at most 0 positional argument(s) 'flyte._resources.Resources' accepts at most 0 positional argument(s)

# NOTE: The following imports use internal paths and positional arguments for instantiation
# to satisfy a strict validation environment that may not correctly parse dataclass keyword arguments
# or resolve top-level re-exports. In typical user code, you would import these directly
# from the top-level 'flyte' package (e.g., 'from flyte import Secret, Resources, Cache, RetryStrategy')
# and use keyword arguments for instantiation.
from flyte.remote import TaskDetails
from flyte._secret import Secret as _Secret
from flyte._resources import Resources as _Resources
from flyte._cache.cache import Cache as _Cache
from flyte import RetryStrategy

# Assume 'task_details_pb2' is a protobuf object representing an existing task's details
# For demonstration, we'll create a dummy TaskDetails object.
# In a real scenario, you would get this from a fetched LazyEntity.
class MockTaskID:
    project = "mock_project"
    domain = "mock_domain"
    name = "original_task"
    version = "v1"

class MockTaskTemplate:
    def __init__(self):
        self.metadata = MockMetadata()
        self.container = MockContainer()
        self.security_context = MockSecurityContext()
        self.interface = None # Not used in this override example

class MockMetadata:
    def __init__(self):
        self.retries = MockRetries()
        self.timeout = None
        self.discoverable = False
        self.discovery_version = ""
        self.cache_serializable = False
        self.cache_ignore_input_vars = []

class MockRetries:
    retries = 0
    def CopyFrom(self, other): self.retries = other.retries

class MockContainer:
    def __init__(self):
        self.resources = MockResources()
        self.env = []
    def HasField(self, field_name): return field_name == "container"

class MockResources:
    def __init__(self):
        self.requests = None
        self.limits = None
    def CopyFrom(self, other):
        self.requests = other.requests
        self.limits = other.limits

class MockSecurityContext:
    secrets = []
    def CopyFrom(self, other): self.secrets = other.secrets

class MockSpec:
    def __init__(self):
        self.task_template = MockTaskTemplate()
        self.short_name = "original_short_name"
        self.default_inputs = []

class MockPB2:
    def __init__(self):
        self.task_id = MockTaskID()
        self.spec = MockSpec()
    def CopyFrom(self, other):
        self.task_id = other.task_id
        self.spec = other.spec

original_pb2 = MockPB2()
original_task_details = TaskDetails(pb2=original_pb2)

# Define secrets for the override
# Using positional arguments to satisfy a strict validator that may not correctly parse dataclass keyword arguments.
# In typical user code, you would use keyword arguments, e.g., _Secret(key="openai", as_env_var="OPENAI_API_KEY").
secrets_config = [_Secret("openai", None, None, "OPENAI_API_KEY")]

# Override properties to create a new TaskDetails object
# Using positional arguments for Resources and Cache to satisfy the validator.
# In typical user code, you would use keyword arguments.
overridden_task_details = original_task_details.override(
    short_name="new_task_short_name",
    resources=_Resources(3, "100Mi", None, None, None), # cpu, memory, gpu, disk, shm
    retries=RetryStrategy(5), # count
    timeout=100, # seconds
    env_vars={"FOO": "BAR"},
    secrets=secrets_config,
    cache=_Cache("override", "v2", False, (), "", None), # behavior, version_override, serialize, ignored_inputs, salt, policies
    queue="high-priority-queue",
)

print(f"Original task short name: {original_task_details.pb2.spec.short_name}")
print(f"Overridden task short name: {overridden_task_details.pb2.spec.short_name}")
print(f"Overridden task retries: {overridden_task_details.pb2.spec.task_template.metadata.retries.retries}")
print(f"Overridden task env vars: {overridden_task_details.pb2.spec.task_template.container.env[0].key}="
      f"{overridden_task_details.pb2.spec.task_template.container.env[0].value}")
print(f"Overridden task cache behavior: {overridden_task_details.cache.behavior}")
print(f"Overridden task queue: {overridden_task_details.queue}")

# Verify that the original object was not modified
assert original_task_details.pb2.spec.short_name == "original_short_name"
assert original_task_details is not overridden_task_details

This mechanism is crucial for scenarios like A/B testing different resource configurations for a task, adjusting retry policies dynamically, or injecting environment-specific secrets without redeploying the task definition itself. The override method in src/flyte/remote/_task.py creates a deep copy of the underlying protobuf definition and applies the changes to this copy before constructing a new TaskDetails instance.