Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, enforce specific input values, and automate runs via schedules. While every workflow automatically receives a default launch plan upon registration, you can define custom launch plans to create specialized execution templates for different environments or recurring tasks.

Parameterizing Workflows with Launch Plans​

When you need to run a workflow with a specific set of default values or lock certain inputs so they cannot be changed at execution time, you use the LaunchPlan class.

Default and Fixed Inputs​

Flytekit distinguishes between default_inputs (which can be overridden by the user at launch time) and fixed_inputs (which are immutable for that specific launch plan).

If you attempt to provide a value for a fixed input during execution, Flyte will reject the request. This is useful for "production" launch plans where certain parameters like environment="prod" should never be altered by a manual trigger.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(name: str, threshold: float, environment: str) -> str:
    return f"Running {name} in {environment} with threshold {threshold}"

# Create a launch plan with some defaults and some fixed values
standard_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="standard_launch_plan",
    default_inputs={"threshold": 0.5},
    fixed_inputs={"environment": "staging"}
)

Internally, LaunchPlan.get_or_create uses the create class method to process these inputs. It converts Python native values into Flyte literals using translate_inputs_to_literals and ensures that fixed inputs are removed from the ParameterMap (the set of inputs a user is allowed to provide).

The Default Launch Plan​

Every workflow has a default launch plan that shares the workflow's name. You can retrieve it using LaunchPlan.get_or_create(workflow).

@workflow
def simple_wf(a: int):
    ...

# This retrieves the auto-generated default launch plan
default_lp = LaunchPlan.get_or_create(simple_wf)

The get_default_launch_plan method in LaunchPlan handles this by extracting the workflow's interface and creating a plan with no fixed inputs and no schedule.

Scheduling Executions​

To automate workflow runs, you attach a schedule to a launch plan. Flytekit supports two primary scheduling mechanisms: CronSchedule and FixedRate.

Cron-Based Schedules​

Use CronSchedule when you need complex timing, such as running a workflow every weekday at a specific hour.

from flytekit import LaunchPlan, CronSchedule

daily_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="daily_report_plan",
    schedule=CronSchedule(
        schedule="0 9 * * 1-5",  # 9 AM UTC, Monday through Friday
        kickoff_time_input_arg="kickoff_time"
    ),
    default_inputs={"name": "daily_report"}
)

The kickoff_time_input_arg parameter is a convenience feature. If your workflow defines an input (e.g., kickoff_time: datetime), Flyte will automatically inject the exact time the schedule triggered the execution into that argument.

Interval-Based Schedules​

Use FixedRate for simple recurring intervals. The minimum supported granularity is one minute.

from datetime import timedelta
from flytekit import LaunchPlan, FixedRate

frequent_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="frequent_check_plan",
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"name": "health_check"}
)

The FixedRate class validates the duration in its __init__ and translates it into a FixedRateUnit (MINUTE, HOUR, or DAY) that the Flyte engine understands.

Referencing Existing Launch Plans​

If you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow or script, use ReferenceLaunchPlan. This allows you to interact with the entity without needing the original source code of the workflow it triggers.

from flytekit import ReferenceLaunchPlan

existing_lp = ReferenceLaunchPlan(
    project="flytesnacks",
    domain="development",
    name="standard_launch_plan",
    version="v1",
    inputs={"name": str, "threshold": float},
    outputs={"o0": str}
)

You can also use the @reference_launch_plan decorator to define the interface more naturally using a Python function signature.

Execution Behavior​

When you call a LaunchPlan object locally, it behaves like the underlying workflow but incorporates the saved_inputs (the combination of default and fixed inputs).

# Local execution uses the defaults/fixed values defined in the LP
result = standard_lp(name="manual_run", threshold=0.8)

In the LaunchPlan.__call__ implementation, flytekit checks the FlyteContext. If it's in a compilation state (e.g., during registration or when used inside another workflow), it creates a node in the workflow graph using create_and_link_node. Otherwise, it simply executes the underlying workflow function with the merged inputs.