Launch plans, schedules, and fixed inputs

Launch plans in flytekit parameterize workflow executions by binding a workflow to specific configurations, such as default or fixed inputs, schedules, and notifications. While every workflow is automatically registered with a default launch plan, you can create custom launch plans to define specialized execution profiles for the same workflow logic.

Parameterizing Workflows

A LaunchPlan acts as a wrapper around a workflow, allowing you to pre-configure how that workflow should be invoked. You create them using the LaunchPlan.get_or_create factory method in flytekit.core.launch_plan.

Default Launch Plans

Every workflow has a default launch plan that inherits the workflow's signature and default values. You can retrieve it using:

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

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

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

Internally, LaunchPlan.get_default_launch_plan derives the parameter map from the workflow's python_interface and caches the result in LaunchPlan.CACHE using the workflow name.

Default vs. Fixed Inputs

When creating a named launch plan, you can distinguish between inputs that can be overridden at launch time and those that are locked.

• Default Inputs: These provide values that the user can change when triggering the execution.
• Fixed Inputs: These are "baked into" the launch plan and cannot be changed at launch time.

named_lp = LaunchPlan.get_or_create(
    name="specialized_plan",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"c": "fixed_value"}
)

In LaunchPlan.create, flytekit removes keys present in fixed_inputs from the parameters map (the set of inputs exposed to the user). These fixed values are stored as a LiteralMap in self.fixed_inputs.

Scheduling Executions

Flytekit provides two primary mechanisms for scheduling launch plans: CronSchedule and FixedRate. These are defined in flytekit.core.schedule.

Cron Schedules

CronSchedule supports standard 5-field cron expressions or predefined aliases like @daily or @hourly.

from flytekit.core.schedule import CronSchedule

# Runs every minute
cron_lp = LaunchPlan.get_or_create(
    name="cron_plan",
    workflow=my_wf,
    schedule=CronSchedule(schedule="*/1 * * * *"),
    default_inputs={"a": 5}
)

You can also pass a kickoff_time_input_arg to inject the scheduled time into a specific workflow input:

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

schedule = CronSchedule(
    schedule="0 0 * * *",
    kickoff_time_input_arg="kickoff_time"
)

Fixed Rate Schedules

FixedRate uses a datetime.timedelta to define a recurring interval. Flytekit supports granularity down to one minute; sub-minute precision will raise an AssertionError.

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

# Runs every 10 minutes
rate_lp = LaunchPlan.get_or_create(
    name="rate_plan",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Internally, FixedRate._translate_duration converts the timedelta into the largest possible unit (days, hours, or minutes) supported by the Flyte IDL.

Execution and Compilation

A LaunchPlan is a callable entity. Its behavior depends on whether it is called during local execution or during workflow compilation.

Local Execution

When you call a launch plan object in a local Python script, it merges the saved_inputs (defaults and fixed values) with any keyword arguments you provide and executes the underlying workflow:

# Executes my_wf(a=10, c="fixed_value") locally
named_lp(a=10)

Note that LaunchPlan.__call__ only accepts keyword arguments.

Workflow Compilation

When a launch plan is called inside another @workflow, flytekit's compilation state is active. Instead of executing the workflow, LaunchPlan.__call__ invokes create_and_link_node, which creates a new node in the workflow graph.

This allows you to use launch plans as building blocks within other workflows, which is particularly useful for dynamic tasks that need to trigger pre-registered launch plans:

from flytekit import dynamic

@dynamic(node_dependency_hints=[named_lp])
def dynamic_wf():
    # Returns a list of nodes executing the launch plan
    return [named_lp(a=i) for i in range(5)]

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster without having the source code for the underlying workflow, use a ReferenceLaunchPlan.

from flytekit.core.launch_plan import reference_launch_plan

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

The @reference_launch_plan decorator uses the function's type hints to construct the interface. Because these refer to external entities, they cannot be executed locally and are used primarily as pointers during registration and compilation.

Implementation Details and Constraints

• Caching: Launch plans are cached by name in LaunchPlan.CACHE. Attempting to create two different launch plans with the same name will raise an AssertionError.
• Security Context: You can specify a security_context for the execution. The older auth_role parameter is deprecated and is internally converted to a SecurityContext. Specifying both will result in a ValueError.
• Auto-Activation: Setting auto_activate=True in get_or_create ensures the schedule is activated immediately upon registration.
• Fixed Input Immutability: While LaunchPlan removes fixed inputs from the public parameter map, the core __call__ implementation merges kwargs over saved_inputs. Users should respect the contract that fixed inputs are not intended to be overridden.