Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, apply fixed or default inputs, and define execution schedules. While every workflow is registered with a default launch plan, you can create custom launch plans to manage different execution scenarios, such as production runs with specific resource labels or recurring maintenance tasks.

Creating Launch Plans

The primary way to create a launch plan is through the LaunchPlan.get_or_create method in flytekit.core.launch_plan. This method ensures that launch plans are cached and reused, preventing duplicate registrations for the same workflow and configuration.

Default Launch Plans

A default launch plan uses the workflow's name and inherits all default values defined in the workflow signature. It does not include additional schedules, notifications, or fixed inputs.

from flytekit import workflow
from flytekit.core.launch_plan import LaunchPlan

@workflow
def my_wf(a: int, c: str) -> str:
    ...

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

Parameterizing with Default and Fixed Inputs

You can customize a launch plan by providing default_inputs and fixed_inputs.

• Default Inputs: These provide values that can be overridden at execution time.
• Fixed Inputs: These values are locked and cannot be changed when the launch plan is invoked.

If a parameter is defined in both default_inputs and fixed_inputs, the LaunchPlan constructor (in flytekit/core/launch_plan.py) ensures the fixed value takes precedence and removes the parameter from the user-facing input map.

from flytekit.core.launch_plan import LaunchPlan

# A named launch plan with specific inputs
production_lp = LaunchPlan.get_or_create(
    name="production_execution",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"c": "production-cluster"}
)

Internally, LaunchPlan.create uses transform_inputs_to_parameters and translate_inputs_to_literals to convert Python native types into Flyte's internal ParameterMap and LiteralMap models.

Scheduling Executions

Flytekit supports recurring executions through the schedule parameter in LaunchPlan.get_or_create. You can define schedules using either cron expressions or fixed intervals.

Cron Schedules

The CronSchedule class in flytekit.core.schedule supports standard cron formats and aliases like @daily or @hourly.

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

daily_lp = LaunchPlan.get_or_create(
    name="daily_sync",
    workflow=my_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
    ),
    default_inputs={"a": 1, "c": "daily"}
)

You can also pass the kickoff_time_input_arg to CronSchedule. This allows your workflow to receive the exact time the execution was triggered as a datetime input.

Fixed Rate Schedules

For simple intervals, use the FixedRate class, which accepts a datetime.timedelta.

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

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

Note that FixedRate schedules (implemented in flytekit/core/schedule.py) do not support granularity of less than one minute and will raise an AssertionError if a smaller duration is provided.

Launch Plans in Dynamic Workflows

When using launch plans inside a @dynamic task, you must provide them as node_dependency_hints. This ensures that flytekit registers the launch plan with Flyte Admin before the dynamic task attempts to execute it.

from flytekit import dynamic, workflow
from flytekit.core.launch_plan import LaunchPlan

@workflow
def sub_wf(x: int):
    ...

lp = LaunchPlan.get_or_create(sub_wf)

@dynamic(node_dependency_hints=[lp])
def launch_dynamically(n: int):
    # The launch plan must be registered on flyteadmin to be called here
    return [lp(x=i) for i in range(n)]

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster without redefining it, use ReferenceLaunchPlan. This class acts as a pointer and requires you to specify the expected interface (inputs and outputs) so that flytekit can perform compilation checks.

from flytekit.core.launch_plan import ReferenceLaunchPlan

existing_lp = ReferenceLaunchPlan(
    project="flytesnacks",
    domain="development",
    name="daily_sync",
    version="v1",
    inputs={"a": int, "c": str},
    outputs={"o0": str}
)

Execution and Call Semantics

When you call a LaunchPlan object, it behaves differently depending on the context:

1. Compilation Context: If called within a workflow or dynamic task, it returns a node via create_and_link_node.
2. Local Execution: If called outside a Flyte context, it forwards the call to the underlying workflow, merging the saved_inputs (defaults and fixed values) with any keyword arguments provided at the call site.

Launch plans only support keyword arguments. Attempting to pass positional arguments will result in an AssertionError.