Launch plans, schedules, and fixed inputs

Launch plans in flytekit allow you to parameterize workflow executions, apply fixed or default inputs, and define schedules for automated runs. While every workflow is registered with a default launch plan, creating custom launch plans enables you to reuse workflow logic with different configurations without modifying the workflow code itself.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method in flytekit/core/launch_plan.py. If you do not provide a name, flytekit returns the default launch plan for the workflow. If you specify additional properties like schedules or fixed inputs, you must provide a unique name.

from flytekit import workflow, LaunchPlan

@workflow
def my_workflow(a: int, b: str) -> str:
    return f"{b}: {a}"

# Get the default launch plan (no custom parameters allowed)
default_lp = LaunchPlan.get_or_create(workflow=my_workflow)

# Create a named launch plan with custom settings
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_launch_plan",
    workflow=my_workflow,
    default_inputs={"a": 10},
    fixed_inputs={"b": "Fixed Value"}
)

Internally, LaunchPlan.get_or_create manages a cache (LaunchPlan.CACHE) to ensure that multiple calls for the same launch plan name return the same object. If you attempt to create a launch plan with an existing name but different parameters, flytekit raises an AssertionError to prevent configuration conflicts.

Parameterizing Inputs

Launch plans distinguish between default inputs and fixed inputs:

1. Default Inputs: These provide values that can be overridden at execution time. They are merged with the workflow's signature defaults, with the launch plan's values taking precedence.
2. Fixed Inputs: These values are locked and cannot be changed when the launch plan is invoked.

In LaunchPlan.create, flytekit processes these inputs by:

• Updating the workflow's ParameterMap with the values provided in default_inputs.
• Translating fixed_inputs into a LiteralMap using translate_inputs_to_literals.
• Removing fixed inputs from the ParameterMap so they are no longer exposed as tunable parameters at launch time.

# 'a' can be changed at runtime, but defaults to 42
# 'b' is locked to "constant" and cannot be changed
lp = LaunchPlan.get_or_create(
    name="input_demo",
    workflow=my_workflow,
    default_inputs={"a": 42},
    fixed_inputs={"b": "constant"}
)

Scheduling Executions

You can automate workflow runs by attaching a schedule to a launch plan. flytekit provides two primary schedule types in flytekit/core/schedule.py: CronSchedule and FixedRate.

Cron Schedules

CronSchedule supports standard cron expressions or aliases (e.g., @daily, @hourly). It is validated using the croniter library.

from flytekit import CronSchedule

daily_lp = LaunchPlan.get_or_create(
    name="daily_sync",
    workflow=my_workflow,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="kickoff_time" # Optional: passes the trigger time to a workflow input
    ),
    default_inputs={"a": 1, "b": "daily"}
)

Fixed Rate Schedules

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

from datetime import timedelta
from flytekit import FixedRate

frequent_lp = LaunchPlan.get_or_create(
    name="every_ten_minutes",
    workflow=my_workflow,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 5, "b": "frequent"}
)

Reference Launch Plans

When you need to trigger a launch plan that is already registered on a Flyte cluster (potentially in a different project or domain), use ReferenceLaunchPlan. This allows you to reference the entity without having the original workflow source code available.

You can define a reference using the ReferenceLaunchPlan class or the @reference_launch_plan decorator:

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my_workflow_lp",
    version="v1"
)
def my_lp_ref(a: int, b: str) -> str:
    ...

The decorator uses transform_function_to_interface to extract the expected types, ensuring that the local reference matches the interface of the remote launch plan during compilation.

Local Execution and Nesting

Launch plans are callable objects. When called locally, they behave like the underlying workflow but inject the saved_inputs (the combination of default and fixed inputs) before execution.

# Locally executes the workflow with a=10 and b="Fixed Value"
result = custom_lp(a=10) 

In a workflow context, calling a launch plan creates a node in the execution graph. The LaunchPlan.__call__ method detects if it is inside a compilation state and uses create_and_link_node to integrate the launch plan as a sub-entity within the workflow.