Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, apply fixed or default inputs, and define schedules or triggers. While every workflow is registered with a default launch plan, you can create named launch plans to customize execution behavior without modifying the underlying workflow logic.

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_wf(a: int, b: str = "default") -> str:
    ...

# Get the default launch plan (no custom name or properties)
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

# Create a named launch plan with custom settings
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_plan",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed_value"}
)

Internally, LaunchPlan.get_or_create caches plans by name in LaunchPlan.CACHE. If you attempt to create a launch plan with an existing name but different properties, flytekit raises an AssertionError to prevent configuration conflicts.

Input Parameterization

Launch plans manage three layers of inputs that determine the final values used during execution:

1. Workflow Defaults: Values defined in the workflow function signature.
2. Default Inputs: Values provided via the default_inputs argument in LaunchPlan.get_or_create. These override the workflow's signature defaults.
3. Fixed Inputs: Values provided via the fixed_inputs argument. These are "locked" and cannot be overridden at launch time.

Fixed vs. Default Inputs

When you define fixed_inputs, flytekit converts them into a LiteralMap and removes them from the launch plan's parameters (the interface exposed to the user at call time). If you try to pass a value for a fixed input when calling the launch plan, flytekit raises a FlyteAssertion.

# This will fail because 'b' is a fixed input
# custom_lp(a=5, b="new_value") 

Local Execution and Compilation

The LaunchPlan.__call__ method handles both local execution and workflow compilation.

• Local Execution: It merges saved_inputs (which includes both defaults and fixed values) with any keyword arguments provided at the call site and executes the underlying workflow.
• Compilation: It uses create_and_link_node to integrate the launch plan into the workflow graph, ensuring that fixed inputs are correctly bound and defaults are applied.

Scheduling and Triggers

Launch plans can be configured to run automatically using schedules. flytekit supports two primary schedule types in flytekit/core/schedule.py: CronSchedule and FixedRate.

Cron Schedules

CronSchedule supports standard five-field cron expressions or aliases like @daily or hourly. It also supports an optional offset (ISO 8601 duration) and a kickoff_time_input_arg to inject the scheduled time into the workflow.

from flytekit import CronSchedule

daily_schedule = CronSchedule(
    schedule="0 0 * * *",  # Runs at midnight every day
    kickoff_time_input_arg="kickoff_time"
)

# The workflow must accept the kickoff_time argument
@workflow
def my_scheduled_wf(kickoff_time: datetime):
    ...

Fixed Rate Schedules

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

from datetime import timedelta
from flytekit import FixedRate

ten_minute_schedule = FixedRate(duration=timedelta(minutes=10))

Applying Schedules

You attach a schedule to a launch plan using the schedule parameter. Alternatively, the trigger parameter (currently in alpha) accepts an OnSchedule object.

scheduled_lp = LaunchPlan.get_or_create(
    name="daily_plan",
    workflow=my_wf,
    schedule=daily_schedule
)

Reference Launch Plans

When you need to trigger a launch plan that is already registered on a Flyte cluster without having the source code available, use ReferenceLaunchPlan. This class serves as a pointer and does not perform network calls during construction.

from flytekit.core.launch_plan import ReferenceLaunchPlan

ref_lp = ReferenceLaunchPlan(
    project="my_project",
    domain="development",
    name="my_registered_lp",
    version="v1",
    inputs={"a": int},
    outputs={"o0": str}
)

Because reference entities do not have access to the original implementation, you must explicitly provide the inputs and outputs types. If the provided interface does not match the actual registered entity, compilation or registration will fail.

Integration in Dynamic Tasks

Launch plans are often used within dynamic tasks to trigger sub-workflows. To ensure the Flyte backend recognizes these dependencies, you must include the launch plan in the node_dependency_hints of the @dynamic decorator.

from flytekit import dynamic

@dynamic(node_dependency_hints=[custom_lp])
def my_dynamic_task(n: int):
    # Returns multiple executions of the launch plan
    return [custom_lp(a=i) for i in range(n)]

This hint ensures that custom_lp is registered on Flyte Admin before the dynamic task attempts to invoke it.