Deploy and Inspect Environments

In flyte-sdk, deploying an environment involves a structured process that separates the definition of what needs to be deployed from the actual execution and the representation of the deployment's outcome. This guide explains how flyte-sdk handles the planning, execution, and inspection of environments and their associated tasks.

Deployment Planning with DeploymentPlan

The initial step in any deployment is defining what environments and tasks are to be deployed. flyte-sdk uses the DeploymentPlan class, found in flyte._deploy.DeploymentPlan, as the blueprint for this process.

The DeploymentPlan is a dataclass that holds a dictionary of environment names mapped to Environment objects, and an optional version string.

from typing import Dict, Optional, TYPE_CHECKING
from dataclasses import dataclass
import rich.repr

if TYPE_CHECKING:
    from flyte._task_environment import Environment

@rich.repr.auto
@dataclass
class DeploymentPlan:
    envs: Dict[str, "Environment"]
    version: Optional[str] = None

A DeploymentPlan is typically constructed with one or more TaskEnvironment objects, which define the specific environments to be deployed. For instance, to plan the deployment of a custom environment and a default environment:

from flyte._deploy import DeploymentPlan
from flyte._task_environment import TaskEnvironment

custom_env_name = "my_custom_env"
default_env_name = "my_default_env"
custom_image = "my_registry/my_image:latest"

custom_env = TaskEnvironment(name=custom_env_name, image=custom_image)
default_env = TaskEnvironment(name=default_env_name)
deployment_plan = DeploymentPlan()
deployment_plan.envs = {custom_env_name: custom_env, default_env_name: default_env}

This deployment_plan then serves as input for subsequent build or deployment operations, such as flyte.build_images.

The Deployment Context (DeploymentContext)

When a deployment operation is initiated, flyte-sdk encapsulates all necessary information into a DeploymentContext object, defined in flyte._deployer.DeploymentContext. This context is passed to the deployer to execute the deployment.

from typing import TYPE_CHECKING
from dataclasses import dataclass
import rich.repr

if TYPE_CHECKING:
    from flyte._task_environment import Environment
    from flyte.core.serialization import SerializationContext

@rich.repr.auto
@dataclass
class DeploymentContext:
    """
    Context for deployment operations.
    """

    environment: "Environment"
    serialization_context: "SerializationContext"
    dryrun: bool = False

The DeploymentContext includes:

• environment: The specific Environment object currently being deployed.
• serialization_context: A SerializationContext used for handling data serialization during the deployment process.
• dryrun: A boolean flag that, when True, indicates that the deployment should be simulated without making actual changes. This is useful for pre-flight checks.

Executing Deployments with Deployer

flyte-sdk defines a Deployer protocol in flyte._deployer.Deployer to abstract the actual deployment logic. This protocol specifies an interface for callables responsible for deploying an environment.

from typing import Protocol, TYPE_CHECKING

if TYPE_CHECKING:
    from flyte._deployer import DeployedEnvironment
    from flyte._deployer import DeploymentContext

class Deployer(Protocol):
    """
    Protocol for deployment callables.
    """

    async def __call__(self, context: DeploymentContext) -> DeployedEnvironment:
        """
        Deploy the environment described in the context.

        Args:
            context: Deployment context containing environment, serialization context, and dryrun flag

        Returns:
            Deployment result
        """
        ...

Implementations of the Deployer protocol take a DeploymentContext and return a DeployedEnvironment, representing the outcome of the deployment for that specific environment.

The Orchestration Layer: flyte.deploy

The central orchestration for deployments in flyte-sdk is handled by the flyte.deploy function (which internally uses an apply function in src/flyte/_deploy.py). This function takes a DeploymentPlan and iterates through each environment within it. For each environment, it creates a DeploymentContext, retrieves the appropriate Deployer implementation, and invokes it to perform the deployment.

import asyncio
from typing import Dict

# Simplified imports for illustration
from flyte._deploy import DeploymentPlan, Deployment
from flyte._deployer import DeploymentContext, get_deployer
from flyte._deployer import DeployedEnvironment # Assuming DeployedEnvironment is imported for type hinting
from flyte.core.serialization import SerializationContext # Assuming SerializationContext is imported

async def _orchestrate_deployment(deployment_plan: DeploymentPlan, sc: SerializationContext, dryrun: bool) -> Deployment:
    deployment_coros = []
    for env_name, env in deployment_plan.envs.items():
        # In a real scenario, status updates would be shown
        deployer = get_deployer(type(env))
        context = DeploymentContext()
        context.environment = env
        context.serialization_context = sc
        context.dryrun = dryrun
        deployment_coros.append(deployer(context))
    deployed_envs = await asyncio.gather(*deployment_coros)
    envs = {}
    for d in deployed_envs:
        envs[d.get_name()] = d

    deployment_result = Deployment()
    deployment_result.envs = envs
    return deployment_result

This process aggregates the results from individual environment deployments into a single Deployment object.

Representing Deployed Entities

flyte-sdk provides a hierarchy of classes to represent the state and details of deployed entities:

DeployedEnvironment

DeployedEnvironment (from flyte._deployer.DeployedEnvironment) is a protocol that defines a generic interface for any deployed environment. It specifies methods for retrieving the environment's name, and various representations for display purposes.

from typing import List, Tuple, Protocol

class DeployedEnvironment(Protocol):
    """
    Protocol for deployed environment representations.
    """

    def get_name(self) -> str:
        """
        Returns the name of the deployed environment.
        """
        ...

    def env_repr(self) -> List[Tuple[str, ...]]:
        """
        Returns a detailed representation of the deployed environment.
        """
        ...

    def table_repr(self) -> List[List[Tuple[str, ...]]]:
        """
        Returns a detailed representation of the deployed entities in the environment, useful for tabular display.
        """
        ...

    def summary_repr(self) -> str:
        """
        Returns a summary representation of the deployed environment.
        """
        ...

DeployedTaskEnvironment

DeployedTaskEnvironment (from flyte._deploy.DeployedTaskEnvironment) is a concrete implementation of DeployedEnvironment that represents a deployed environment containing tasks. It holds the original TaskEnvironment and a list of DeployedTask objects.

from typing import List, Tuple, TYPE_CHECKING
from dataclasses import dataclass
import rich.repr

if TYPE_CHECKING:
    from flyte._task_environment import TaskEnvironment
    from flyte._deploy import DeployedTask
    from flyte._task_environment import Image # For env_repr method

@rich.repr.auto
@dataclass
class DeployedTaskEnvironment:
    env: "TaskEnvironment"
    deployed_entities: List["DeployedTask"]

    def get_name(self) -> str:
        """
        Returns the name of the deployed environment.
        """
        return self.env.name

    def summary_repr(self) -> str:
        """
        Returns a summary representation of the deployment.
        """
        entities = ", ".join(f"{e.summary_repr()}" for e in self.deployed_entities or [])
        return f"Deployment(env=[{self.env.name}], entities=[{entities}])"

    def table_repr(self) -> List[List[Tuple[str, ...]]]:
        """
        Returns a detailed representation of the deployed tasks.
        """
        tuples = []
        if self.deployed_entities:
            for e in self.deployed_entities:
                tuples.append(e.table_repr())
        return tuples

    def env_repr(self) -> List[Tuple[str, ...]]:
        """
        Returns a detailed representation of the deployed environments.
        """
        env = self.env
        return [
            ("environment", env.name),
            ("image", env.image.uri if isinstance(env.image, Image) else env.image or ""),
        ]

DeployedTaskEnvironment is instantiated after tasks within an environment have been deployed:

import asyncio
from typing import List

# Simplified imports for illustration
from flyte._deploy import DeployedTaskEnvironment, DeployedTask
from flyte._task_environment import TaskEnvironment

async def _deploy_task_environment_example(env: TaskEnvironment, task_coros) -> DeployedTaskEnvironment:
    deployed_task_vals = await asyncio.gather(*task_coros)
    deployed_tasks = []
    for t in deployed_task_vals:
        deployed_tasks.append(t)
    # The DeployedTaskEnvironment constructor does not accept 'env' or 'deployed_entities' as keyword arguments.
    # Instead, these attributes are set directly on the instance after creation.
    deployed_env = DeployedTaskEnvironment()
    deployed_env.env = env
    deployed_env.deployed_entities = deployed_tasks
    return deployed_env

DeployedTask

DeployedTask (from flyte._deploy.DeployedTask) represents an individual task that has been deployed. It contains the task's specification and any associated triggers.

from typing import List, Tuple
from dataclasses import dataclass
import rich.repr

# Assuming task_definition_pb2 is available
from flytekit.models.task import task_definition_pb2

@rich.repr.auto
@dataclass
class DeployedTask:
    deployed_task: task_definition_pb2.TaskSpec
    deployed_triggers: List[task_definition_pb2.TaskTrigger]

    def get_name(self) -> str:
        """
        Returns the name of the deployed environment.
        """
        return self.deployed_task.task_template.id.name

    def summary_repr(self) -> str:
        """
        Returns a summary representation of the deployed task.
        """
        return (
            f"DeployedTask(name={self.deployed_task.task_template.id.name}, "
            f"version={self.deployed_task.task_template.id.version})"
        )

    def table_repr(self) -> List[Tuple[str, ...]]:
        """
        Returns a table representation of the deployed task.
        """
        from flyte._initialize import get_client

        client = get_client()
        task_id = self.deployed_task.task_template.id
        task_url = client.console.task_url(
            project=task_id.project,
            domain=task_id.domain,
            task_name=task_id.name,
        )
        triggers = []
        for t in self.deployed_triggers:
            trigger_url = client.console.trigger_url(
                project=task_id.project,
                domain=task_id.domain,
                task_name=task_id.name,
                trigger_name=t.name,
            )
            triggers.append(f"[link={trigger_url}]{t.name}[/link]")

        return [
            ("type", "task"),
            ("name", f"[link={task_url}]{task_id.name}[/link]"),
            ("version", task_id.version),
            ("triggers", ",".join(triggers)),
        ]

DeployedTask instances are created during the task deployment process, including during dry runs or when a task already exists (idempotency):

# Example of DeployedTask instantiation during a dry run
from flyte._deploy import DeployedTask
# from flytekit.models.task import task_definition_pb2 # Assuming this is available
# from flyte.core.serialization import SerializationContext # Assuming this is available

# def _deploy_task(task, serialization_context, dryrun): # Simplified function signature
#     if dryrun:
#         return DeployedTask(translate_task_to_wire(task, serialization_context), [])

# Example of DeployedTask instantiation for an existing task
from flyte._deploy import DeployedTask
# from flytekit.models.task import task_definition_pb2 # Assuming this is available
# from connectrpc import Code, ConnectError # Assuming these are available

# def _deploy_task_existing(task, spec, deployable_triggers): # Simplified function signature
#     try:
#         # Attempt to deploy task
#         pass
#     except ConnectError as e:
#         if e.code == Code.ALREADY_EXISTS:
#             # status.info(f"Task {task.name} already exists, skipping") # Simplified
#             return DeployedTask(spec, deployable_triggers)
#         raise
#     return DeployedTask(spec, deployable_triggers)

The Final Deployment Result (Deployment)

The Deployment class (from flyte._deploy.Deployment) aggregates all DeployedEnvironment objects, providing a comprehensive view of the entire deployment's outcome. It is a frozen dataclass, ensuring immutability of the deployment results.

from typing import Dict, List, Tuple, TYPE_CHECKING
from dataclasses import dataclass
import rich.repr

if TYPE_CHECKING:
    from flyte._deployer import DeployedEnvironment

@rich.repr.auto
@dataclass(frozen=True)
class Deployment:
    envs: Dict[str, "DeployedEnvironment"]

    def summary_repr(self) -> str:
        """
        Returns a summary representation of the deployment.
        """
        envs = ", ".join(f"{e.summary_repr()}" for e in self.envs.values() or [])
        return f"Deployment(envs=[{envs}])"

    def table_repr(self) -> List[List[Tuple[str, ...]]]:
        """
        Returns a detailed representation of the deployed tasks.
        """
        tuples = []
        for d in self.envs.values():
            tuples.extend(d.table_repr())
        return tuples

    def env_repr(self) -> List[List[Tuple[str, ...]]]:
        """
        Returns a detailed representation of the deployed environments.
        """
        tuples = []
        for d in self.envs.values():
            tuples.append(d.env_repr())
        return tuples

Key Features and Gotchas

• Dry Run Capability: The dryrun flag in DeploymentContext allows you to simulate a deployment without making any actual changes to the remote Flyte environment. This is invaluable for validating your deployment configuration and catching potential issues early.
• Idempotency: flyte-sdk handles existing tasks gracefully. If a task with the same identifier already exists during deployment, it will be skipped, as indicated by the ConnectError with Code.ALREADY_EXISTS being caught and an existing DeployedTask being returned.
• Internal Packages: It's important to note that packages like flyte._deploy, flyte._deployer, flyte._initialize, flyte.cli._build, and flyte.cli._deploy are considered internal to the flyte-sdk. While this documentation explains their mechanics, users are generally encouraged to interact with flyte-sdk through its higher-level APIs or the command-line interface rather than directly importing and using these private modules, as their internal structure may change without public notice.

CLI Workflow for Building and Deploying Environments

flyte-sdk provides command-line tools to streamline the process of building and deploying environments.

Building Environments via CLI

The flyte build command is used to build images for your environments. This command leverages flyte.cli._build.BuildArguments to define its options.

To build images for environments defined in your project, you would typically run:

flyte build <environment_file.py> <environment_name>

For example, if you have an environment named my_env in my_environments.py:

flyte build my_environments.py my_env

You can specify a copy_style or override the root_dir:

flyte build --copy-style loaded_modules --root-dir /path/to/my/project my_environments.py my_env

The BuildEnvCommand class in flyte.cli._build orchestrates this process, calling flyte.build_images internally.

Deploying Environments via CLI

The flyte deploy command facilitates the deployment of environments to a Flyte cluster. This command uses flyte.cli._deploy.DeployArguments to configure deployment options.

To deploy a single environment, you specify the file and the environment name:

flyte deploy <environment_file.py> <environment_name>

For example:

flyte deploy my_environments.py my_env

Performing a Dry Run

To simulate a deployment without making actual changes, use the --dry-run flag:

flyte deploy --dry-run my_environments.py my_env

Deploying All Environments in a File or Directory

To deploy all environments found within a specific Python file, or recursively within a directory, you can use the --all or --recursive flags. The DeployEnvRecursiveCommand in flyte.cli._deploy handles these scenarios.

Deploy all environments in a file:

flyte deploy --all my_environments.py

Deploy all environments recursively in the current directory:

flyte deploy --recursive .

Specifying Version and Images

You can specify a particular version for the deployment or provide custom images:

flyte deploy --version v1.0.0 --image my_image=my_registry/my_image:latest my_environments.py my_env

Other useful options include --project, --domain, --ignore-load-errors (to continue deployment even if some environments fail to load), and --no-sync-local-sys-paths (to disable synchronization of local system paths to the remote container).