Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, apply fixed or default inputs, and attach schedules or notifications. While every workflow is registered with a default launch plan, you can create named launch plans to define specific execution configurations that can be triggered manually or automatically.

Parameterizing Workflows with Launch Plans

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

Default and Fixed Inputs

You can define default_inputs which act as overrides for the workflow's function signature defaults, and fixed_inputs which are immutable at the time of execution.

from flytekit import workflow, LaunchPlan

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

# Create a launch plan with a new default for 'b' and a fixed value for 'a'
my_lp = LaunchPlan.get_or_create(
    name="fixed_a_lp",
    workflow=my_workflow,
    default_inputs={"b": "new default"},
    fixed_inputs={"a": 42}
)

Internally, LaunchPlan.create (called by get_or_create) processes these inputs:

1. It merges workflow signature defaults with default_inputs, where the latter takes precedence.
2. It translates fixed_inputs into Flyte literals using translate_inputs_to_literals.
3. It removes any keys present in fixed_inputs from the ParameterMap stored in self._parameters. This ensures that the Flyte platform does not prompt for or allow changes to these values during launch.
4. It stores the original Python native values in _saved_inputs to support local execution and compilation.

Local and Compiled Execution

When you call a launch plan object (e.g., my_lp(b="override")), flytekit handles it differently based on the context:

• During Compilation: LaunchPlan.__call__ uses create_and_link_node to integrate the launch plan into a workflow graph. It combines the saved_inputs with any keyword arguments provided at the call site.
• During Local Execution: It forwards the call to the underlying workflow, merging the saved_inputs with the call-site arguments.

Note that LaunchPlan only supports keyword arguments; passing positional arguments will raise an AssertionError.

Scheduling Executions

Flytekit allows you to automate workflow runs by attaching a schedule to a launch plan. Schedules are implemented in the schedule package and support both fixed intervals and cron expressions.

Cron Schedules

The CronSchedule class supports standard 5-field cron formats or aliases (like @daily).

from flytekit import LaunchPlan
from flytekit.core.schedule import CronSchedule

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

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

The kickoff_time_input_arg parameter allows you to bind the time the scheduler triggered the execution to a specific workflow input. CronSchedule validates the expression using the croniter library.

Fixed Rate Schedules

For simple periodic execution, use FixedRate. It accepts a datetime.timedelta and enforces a minimum granularity of one minute.

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

# Runs every 10 minutes
ten_min_lp = LaunchPlan.get_or_create(
    name="ten_min_lp",
    workflow=my_workflow,
    schedule=FixedRate(duration=timedelta(minutes=10))
)

FixedRate._translate_duration automatically converts the timedelta into the largest possible unit (Days, Hours, or Minutes) supported by the Flyte backend.

Launch Plan Management and Caching

Flytekit manages launch plans through a process-global cache (LaunchPlan.CACHE).

• Default Launch Plans: If you call LaunchPlan.get_or_create(workflow=wf) without a name, flytekit returns the default launch plan. This plan uses the workflow's name and inherits its default labels and annotations.
• Named Launch Plans: When you provide a name, flytekit ensures uniqueness. If you attempt to create a second launch plan with the same name but different parameters (like a different schedule or different fixed inputs), flytekit raises an AssertionError to prevent configuration drift.

Cloning Launch Plans

If you need to derive a new launch plan from an existing one, use clone_with. This method creates a new instance while allowing you to override specific attributes:

# Create a new launch plan based on an existing one but with a different schedule
new_lp = my_lp.clone_with(
    name="new_scheduled_lp",
    schedule=FixedRate(duration=timedelta(hours=1))
)

Reference Launch Plans

When a launch plan is already registered on a Flyte cluster and you want to trigger it from another workflow without redefining it, use a ReferenceLaunchPlan.

from flytekit import reference_launch_plan

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

The @reference_launch_plan decorator uses transform_function_to_interface with is_reference_entity=True to derive the expected interface from the function signature. This allows flytekit to compile workflows that depend on the reference without needing to fetch the interface from the Flyte Admin service at development time.