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 define specific execution configurations without modifying the underlying workflow code.

Creating Launch Plans

The most common way to interact with launch plans is through the LaunchPlan.get_or_create method. If you do not provide a name, flytekit returns the default launch plan for that workflow.

from flytekit import workflow, LaunchPlan

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

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

Internally, LaunchPlan.get_or_create manages a cache (LaunchPlan.CACHE) to ensure that multiple calls for the same workflow or named launch plan return the same object, preventing redundant entity creation.

Parameterizing with Default and Fixed Inputs

Launch plans allow you to pre-configure inputs in two ways:

1. Default Inputs: Values that can be overridden at execution time.
2. Fixed Inputs: Values that are locked and cannot be changed when the launch plan is invoked.

When you define a launch plan with these inputs, flytekit uses transform_inputs_to_parameters and translate_inputs_to_literals to convert Python types into Flyte's internal IDL models.

# Create a named launch plan with specific inputs
custom_lp = LaunchPlan.get_or_create(
    name="frequent_execution_lp",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed-value"}
)

In the LaunchPlan.create method, flytekit ensures that any key present in fixed_inputs is removed from the parameters map (which holds defaults), ensuring that fixed values remain immutable during the launch process.

Scheduling Workflows

Flytekit supports automated execution through schedules. You can attach a schedule to a launch plan using either CronSchedule or FixedRate.

Cron Schedules

CronSchedule supports standard cron expressions or aliases (like @daily or @hourly). It also allows you to pass the scheduled time into the workflow using the kickoff_time_input_arg.

from flytekit import CronSchedule
from datetime import datetime

@workflow
def scheduled_wf(kickoff_time: datetime):
    print(f"Running for time: {kickoff_time}")

daily_lp = LaunchPlan.get_or_create(
    name="daily_report",
    workflow=scheduled_wf,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="kickoff_time"
    )
)

The CronSchedule class validates the expression using croniter. Note that flytekit distinguishes between cron_expression (deprecated AWS-style 6-field cron) and schedule (standard 5-field cron).

Fixed Rate Schedules

FixedRate is used for intervals defined by a timedelta. Flytekit restricts the granularity of these schedules to a minimum of one minute.

from flytekit import FixedRate
from datetime import timedelta

heartbeat_lp = LaunchPlan.get_or_create(
    name="heartbeat_lp",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 1, "b": "heartbeat"}
)

The FixedRate class internally translates the timedelta into the largest possible unit (Days, Hours, or Minutes) supported by the Flyte backend via the _translate_duration static method.

Execution and Local Behavior

When you call a LaunchPlan object directly in Python, it behaves differently depending on the context:

1. During Compilation: If ctx.compilation_state is active, calling the launch plan creates a node in the workflow graph using create_and_link_node.
2. Local Execution: If called outside of a compilation context, it simply forwards the call to the underlying workflow, merging the saved_inputs (defaults and fixed values) with any keyword arguments provided at the call site.

# Local execution uses the saved inputs from the launch plan
result = custom_lp(a=20) # 'b' is already fixed as "fixed-value"

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster from a different project or domain, use ReferenceLaunchPlan. This class acts as a pointer and requires you to define the expected interface manually, as it does not perform a network call to fetch the interface during instantiation.

from flytekit import ReferenceLaunchPlan

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