Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, define fixed or default inputs, and establish recurring schedules. While every workflow is registered with a default launch plan, you can create custom launch plans to handle specific operational requirements, such as nightly batch runs or executions with pre-configured security contexts.

Creating Launch Plans​

The primary way to interact with launch plans is through the LaunchPlan.get_or_create method in flytekit/core/launch_plan.py. This method ensures that launch plans are cached and reused, preventing duplicate definitions for the same workflow.

Default Launch Plans​

If you do not provide a name, flytekit assumes you want the default launch plan for the workflow. A default launch plan uses the workflow's signature to define its parameters and inherits any default values specified in the @workflow function.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str = "default"):
    ...

# Creates or retrieves the default launch plan named "my_wf"
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

Named Launch Plans with Custom Inputs​

To customize a launch plan with specific inputs or schedules, you must provide a unique name. You can define two types of inputs:

  • default_inputs: Values that can be overridden at execution time.
  • fixed_inputs: Values that are locked and cannot be changed when the launch plan is triggered.

Internally, LaunchPlan.create (called by get_or_create) uses translate_inputs_to_literals to convert these Python values into Flyte's internal LiteralMap format.

# Create a launch plan with one fixed input and one new default
custom_lp = LaunchPlan.get_or_create(
    name="nightly_execution",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed_value"}
)

Scheduling Executions​

Flytekit supports recurring executions through the schedule package. You can define schedules using cron expressions or fixed intervals.

Cron Schedules​

CronSchedule supports standard 5-field cron formats or aliases like @daily and @hourly. It also supports a kickoff_time_input_arg, which allows the workflow to receive the exact time the schedule triggered the execution.

from flytekit import workflow
from flytekit.core.schedule import CronSchedule
from datetime import datetime

@workflow
def scheduled_wf(kickoff_time: datetime):
    ...

daily_lp = LaunchPlan.get_or_create(
    name="daily_lp",
    workflow=scheduled_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",
        kickoff_time_input_arg="kickoff_time"
    )
)

Fixed Rate Schedules​

FixedRate schedules trigger executions at a specific frequency defined by a datetime.timedelta. The minimum supported granularity is one minute.

from datetime import timedelta
from flytekit.core.schedule import FixedRate

ten_minute_lp = LaunchPlan.get_or_create(
    name="ten_minute_lp",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Advanced Configuration​

Launch plans also serve as the container for execution-level metadata and security settings.

Security and Roles​

You can specify a security_context to define the IAM role or Kubernetes service account the workflow should use. Note that auth_role is deprecated in favor of security_context.

from flytekit.models.security import SecurityContext, Identity

secure_lp = LaunchPlan.get_or_create(
    name="secure_lp",
    workflow=my_wf,
    security_context=SecurityContext(
        run_as=Identity(k8s_service_account="my-sva")
    )
)

Notifications and Labels​

Launch plans can be configured to send notifications (e.g., email or Slack) on execution completion or failure, and can attach custom labels and annotations to the resulting executions.

from flytekit.models.common import Labels, Annotations

metadata_lp = LaunchPlan.get_or_create(
    name="metadata_lp",
    workflow=my_wf,
    labels=Labels({"team": "data-science"}),
    annotations=Annotations({"project": "alpha"})
)

Reference Launch Plans​

When you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow, use ReferenceLaunchPlan or the @reference_launch_plan decorator. This allows you to reference the entity by its project, domain, name, and version without needing the original source code.

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="nightly_execution",
    version="v1"
)
def remote_lp(a: int, b: str):
    ...

When a LaunchPlan is called locally (e.g., lp(a=1)), it forwards the call to the underlying workflow. During compilation (e.g., when used as a node in another workflow), LaunchPlan.__call__ uses create_and_link_node to integrate the launch plan into the workflow graph.