Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a way to parameterize workflow executions, apply default or fixed inputs, and define schedules for automatic triggering. While every workflow is registered with a default launch plan, you can create custom ones to handle specific operational requirements like recurring jobs or pre-configured environment settings.

Creating Launch Plans

When you define a workflow, flytekit automatically generates a default launch plan. However, to customize inputs or schedules, you must create a named launch plan using LaunchPlan.get_or_create.

from flytekit import workflow, LaunchPlan

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

# Create a custom launch plan
my_lp = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="parameterized_execution",
    default_inputs={"a": 10, "b": "default_string"}
)

Internally, LaunchPlan.get_or_create (found in flytekit/core/launch_plan.py) manages a cache of launch plans to prevent duplicate creation. If you provide a name, it must be unique within the project and domain. If no name is provided, flytekit returns the default launch plan for that workflow.

Default vs. Fixed Inputs

Launch plans distinguish between inputs that can be overridden at execution time and those that are locked.

• Default Inputs: Use the default_inputs parameter. These values are used if the caller does not provide an alternative.
• Fixed Inputs: Use the fixed_inputs parameter. These values cannot be changed at launch time. If a user attempts to provide a different value for a fixed input during execution, Flyte will reject the request.

fixed_lp = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="fixed_input_plan",
    fixed_inputs={"a": 42},  # 'a' is now locked to 42
    default_inputs={"b": "hello"} # 'b' can still be overridden
)

The LaunchPlan constructor ensures that any key present in fixed_inputs is removed from the parameters map (which represents the user-facing interface), effectively hiding it from the execution trigger interface.

Scheduling Executions

You can automate workflow runs by attaching a schedule to a launch plan. flytekit supports two primary types of schedules: CronSchedule and FixedRate.

Cron Schedules

CronSchedule allows you to define complex recurring patterns using standard cron syntax or aliases like @daily.

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

daily_lp = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="daily_cron_plan",
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
    )
)

The CronSchedule class (in flytekit/core/schedule.py) validates the expression using croniter. It also supports a kickoff_time_input_arg, which allows you to pass the scheduled time into a specific workflow input.

Fixed Rate Schedules

FixedRate is used for simple intervals, such as "every 10 minutes".

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

interval_lp = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="interval_plan",
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Note that FixedRate only supports granularities of one minute or greater. The _translate_duration method in FixedRate automatically converts your timedelta into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY) required by the Flyte IDL.

Launch Plans in Dynamic Workflows

If you need to trigger a launch plan from within a @dynamic task, you must explicitly inform flytekit about the dependency. Because dynamic tasks are compiled at runtime, Flyte needs to know which entities must be registered beforehand.

Use the node_dependency_hints parameter in the @dynamic decorator to include your launch plan:

from flytekit import dynamic, workflow, LaunchPlan

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

sub_lp = LaunchPlan.get_or_create(sub_workflow, name="sub_lp")

@dynamic(node_dependency_hints=[sub_lp])
def dynamic_launcher(count: int):
    # Without node_dependency_hints, this call would fail at runtime
    return [sub_lp(x=i) for i in range(count)]

When a launch plan is called inside a compilation context (like a dynamic task or another workflow), it invokes create_and_link_node to generate a node in the execution graph rather than executing the workflow immediately.

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster but is not defined in your current local codebase, use ReferenceLaunchPlan or the @reference_launch_plan decorator.

from flytekit import reference_launch_plan

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

This creates a pointer to the remote entity without requiring the underlying workflow code. The function signature you provide must match the interface of the remote launch plan exactly, as flytekit uses this signature for local compilation and type checking.