Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, define schedules, and set fixed or default inputs. While every workflow is registered with a default launch plan, you can create custom launch plans to handle specific execution scenarios, such as recurring reports or production runs with locked parameters.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. 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_workflow(a: int, b: str) -> str:
    return f"{b}: {a}"

# Get the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=my_workflow)

# Create a named launch plan with custom settings
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_lp",
    workflow=my_workflow,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed-value"}
)

Internally, LaunchPlan.get_or_create manages a cache (LaunchPlan.CACHE) to ensure that multiple calls for the same launch plan name return the same object. If you attempt to create two launch plans with the same name but different configurations, flytekit raises an AssertionError.

Parameterizing Inputs

Launch plans distinguish between default inputs and fixed inputs:

1. Default Inputs: These provide values that are used if the caller does not provide them. They can be overridden at execution time.
2. Fixed Inputs: These values are locked into the launch plan. They cannot be changed when triggering an execution through that specific launch plan.

When you define fixed_inputs in LaunchPlan.create, flytekit translates these native Python values into LiteralMap objects using translate_inputs_to_literals. It also ensures that fixed inputs are removed from the ParameterMap (the set of inputs exposed to the user) so they cannot be overridden.

# 'a' can be overridden, but 'b' is locked to "constant"
lp = LaunchPlan.get_or_create(
    name="locked_lp",
    workflow=my_workflow,
    default_inputs={"a": 5},
    fixed_inputs={"b": "constant"}
)

Scheduling Executions

Flytekit supports automated execution of launch plans through schedules. You can define these using CronSchedule or FixedRate.

Cron Schedules

CronSchedule allows you to use standard cron expressions or aliases (like @daily or @hourly). Note that flytekit validates these expressions using croniter.

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

daily_lp = LaunchPlan.get_or_create(
    name="daily_report",
    workflow=my_workflow,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="b" # Passes the schedule time to input 'b'
    ),
    default_inputs={"a": 1}
)

Fixed Rate Schedules

FixedRate is used for intervals. The minimum supported granularity is one minute. Flytekit's FixedRate._translate_duration method automatically converts timedelta objects into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY).

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

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

Reference Launch Plans

When you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow or dynamic task, use ReferenceLaunchPlan. This acts as a pointer and does not require the full workflow implementation to be present in the local environment.

You can define a reference using the reference_launch_plan decorator:

from flytekit import reference_launch_plan

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

The decorator uses transform_function_to_interface to extract the expected types, ensuring that the local call site matches the interface of the remote launch plan.

Local and Dynamic Execution

Launch plans are callable objects. When called locally (outside of a compilation context), they simply forward the call to the underlying workflow while merging the saved_inputs (defaults and fixed values).

In a @dynamic task, launch plans must be declared in node_dependency_hints to ensure they are correctly registered and linked on the Flyte backend:

from flytekit import dynamic

@dynamic(node_dependency_hints=[custom_lp])
def launch_dynamically():
    # This creates a node in the dynamic workflow pointing to the launch plan
    return [custom_lp(a=i) for i in range(5)]