Launch plans, schedules, and fixed inputs

Launch plans in flytekit are execution descriptors that wrap workflows to provide pre-defined inputs, schedules, and notifications. While every workflow is registered with a default launch plan, you can create custom launch plans to parameterize executions for different environments or recurring schedules.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. This method ensures that launch plans are cached process-locally, preventing redundant entity creation.

Default Launch Plans

If you do not provide a name, flytekit assumes you want the default launch plan for the workflow. This plan uses the workflow's signature defaults and contains no extra configurations like schedules.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str = "default"):
    ...

# Retrieves or creates the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

Parameterizing with Default and Fixed Inputs

Named launch plans allow you to specialize a workflow's interface. You can provide default_inputs (which can be overridden at launch time) and fixed_inputs (which are locked and cannot be changed).

# Named launch plan with specialized inputs
specialized_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="production_config",
    default_inputs={"a": 10},      # Can be overridden
    fixed_inputs={"b": "locked"}   # Cannot be changed
)

Internally, LaunchPlan.create handles the translation of these Python values:

1. Precedence: Explicit default_inputs override any default values defined in the workflow's Python function signature.
2. Fixed Inputs: Values in fixed_inputs are converted to Flyte Literals and removed from the ParameterMap of the launch plan. This ensures they are not visible as tunable parameters in the Flyte UI or CLI.
3. Validation: If you attempt to create two launch plans with the same name but different properties (like different schedules or inputs), flytekit raises an AssertionError.

Scheduling Executions

Flytekit provides native support for recurring executions through CronSchedule and FixedRate objects. These are attached to a launch plan via the schedule parameter.

Cron Schedules

CronSchedule supports standard 5-field cron expressions and common aliases. It also allows you to pass the "kickoff time" into the workflow as an input.

from flytekit import CronSchedule

# Runs every minute and passes the execution time to the 'kickoff_time' input
cron_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="hourly_cleanup",
    schedule=CronSchedule(
        schedule="*/1 * * * *",
        kickoff_time_input_arg="kickoff_time"
    )
)

Supported aliases include @hourly, @daily, @weekly, @monthly, and @yearly. Note that the 6-field AWS-style cron_expression is deprecated in favor of the 5-field schedule string.

Fixed Rate Intervals

FixedRate uses a datetime.timedelta to define frequency. Flytekit normalizes these durations to the largest possible unit (Day, Hour, or Minute).

from datetime import timedelta
from flytekit import FixedRate

# Runs every 2 hours
rate_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="frequent_sync",
    schedule=FixedRate(duration=timedelta(hours=2))
)

Flytekit enforces a minimum granularity of one minute. If you provide a duration with seconds or microseconds, FixedRate will raise an AssertionError.

Using Launch Plans in Workflows

Launch plans are callable entities. When you call a launch plan inside a workflow, flytekit creates a node that references the launch plan rather than the underlying workflow directly.

@workflow
def parent_wf(val: int):
    # This creates a workflow node pointing to 'specialized_lp'
    return specialized_lp(a=val)

During compilation, LaunchPlan.__call__ invokes create_and_link_node. This ensures that the resulting Flyte IDL contains a launchplan_ref with the correct name and resource type.

Dynamic Execution

If you need to invoke a launch plan from within a @dynamic task, you must declare it in the node_dependency_hints. This informs the Flyte compiler that the launch plan must be registered and available before the dynamic task runs.

from flytekit import dynamic

@dynamic(node_dependency_hints=[specialized_lp])
def dynamic_launcher(n: int):
    return [specialized_lp(a=i) for i in range(n)]

Reference Launch Plans

When you need to trigger a launch plan that is already registered on a Flyte cluster (perhaps in a different project or domain), use a ReferenceLaunchPlan. These act as pointers and do not require the workflow source code.

You can define them using the @reference_launch_plan decorator:

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="core.control_flow.run_merge_sort",
    version="v1"
)
def remote_lp(inputs: list[int]) -> list[int]:
    ...

Because reference entities do not have a local implementation, they cannot be executed locally. You must mock them in your tests using unittest.mock.patch or the flytekit patch utility.

Implementation Details and Constraints

Feature	Behavior
Caching	LaunchPlan.CACHE is keyed by name. Re-creating a plan with the same name but different default_inputs, schedule, or security_context will trigger an AssertionError.
Fixed Inputs	Once an input is moved to fixed_inputs, it is removed from the parameters map. It cannot be overridden during a manual execution.
Auth & Security	The legacy auth_role is automatically converted to a SecurityContext internally. You cannot specify both.
Cloning	LaunchPlan.clone_with allows creating new plans based on existing ones, but it uses truthiness for overrides (e.g., parameters or self.parameters), which can make it difficult to clear certain fields.
Auto-Activation	Setting auto_activate=True in get_or_create ensures the schedule is active immediately upon registration.