Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, define fixed or default inputs, and automate runs via schedules. While every workflow is registered with a default launch plan, creating custom launch plans allows you to lock down specific configurations for different environments or recurring tasks.

Parameterizing Workflows

A LaunchPlan acts as a template for workflow execution. It allows you to specify which inputs are required, which have defaults, and which are fixed.

Default vs. Fixed Inputs

When you define a launch plan using LaunchPlan.get_or_create, you can distinguish between inputs that can be overridden at execution time and those that are immutable:

• default_inputs: These values are used if no input is provided during execution. They can be overridden by the user or a triggering system.
• fixed_inputs: These values are "baked into" the launch plan. They cannot be changed at execution time. If a user attempts to provide a value for a fixed input, flytekit will raise an error.

Internally, LaunchPlan.create handles this by updating the ParameterMap (for defaults) and populating a LiteralMap for fixed_inputs. The constructor ensures that any key present in fixed_inputs is removed from the parameters map to prevent conflicts:

# From flytekit/core/launch_plan.py
# ...
# Ensure fixed inputs are not in parameter map
# parameters = {k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals}
# self._parameters = _interface_models.ParameterMap(parameters=parameters)
# self._fixed_inputs = fixed_inputs

Creating a Launch Plan

To create a launch plan, use the LaunchPlan.get_or_create method. If you only provide the workflow, flytekit returns the default launch plan. If you provide additional attributes like inputs or schedules, you must provide a unique name.

from flytekit import workflow, LaunchPlan

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

# Default launch plan (no name required)
default_lp = LaunchPlan.get_or_create(workflow=my_workflow)

# Custom launch plan with fixed and default inputs
production_lp = LaunchPlan.get_or_create(
    name="prod_launch_plan",
    workflow=my_workflow,
    default_inputs={"a": 42},
    fixed_inputs={"b": "production-run"}
)

Scheduling Executions

Flytekit supports automated workflow execution through schedules. You can define schedules based on fixed time intervals or cron expressions.

Cron Schedules

CronSchedule allows you to trigger workflows using standard cron syntax. Note that the schedule argument is preferred over the deprecated cron_expression.

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

@workflow
def daily_job(report_date: str):
    ...

daily_lp = LaunchPlan.get_or_create(
    name="daily_report_gen",
    workflow=daily_job,
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
    ),
    default_inputs={"report_date": "today"}
)

Fixed Rate Schedules

FixedRate schedules trigger workflows at a consistent frequency defined by a datetime.timedelta. The minimum supported granularity is one minute.

from datetime import timedelta
from flytekit import workflow, LaunchPlan
from flytekit.core.schedule import FixedRate

@workflow
def heartbeat():
    ...

heartbeat_lp = LaunchPlan.get_or_create(
    name="system_heartbeat",
    workflow=heartbeat,
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Capturing Kickoff Time

Often, a scheduled workflow needs to know exactly when it was triggered (e.g., to process data for a specific time window). Both CronSchedule and FixedRate support the kickoff_time_input_arg parameter. This maps the scheduled time to a specific workflow input.

from datetime import datetime
from flytekit import workflow, LaunchPlan
from flytekit.core.schedule import CronSchedule

@workflow
def process_data(kickoff_time: datetime):
    print(f"Execution triggered at {kickoff_time}")

scheduled_lp = LaunchPlan.get_or_create(
    name="time_aware_lp",
    workflow=process_data,
    schedule=CronSchedule(
        schedule="*/5 * * * *",
        kickoff_time_input_arg="kickoff_time"
    )
)

Advanced Triggers and Configuration

Flytekit is evolving its scheduling API. While the schedule argument is standard, the trigger argument (using OnSchedule) is available as an alpha feature for more complex triggering logic.

Notifications and Metadata

Launch plans also serve as the point of configuration for execution-level metadata:

• Notifications: Use the notifications argument to send alerts (Email, Slack, PagerDuty) on execution success or failure.
• Labels and Annotations: Attach K8s metadata to the resulting executions via labels and annotations.
• Security Context: Define the security_context (IAM roles or K8s service accounts) that the workflow should assume.

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster from within another workflow, use ReferenceLaunchPlan. This allows you to reference the entity by its project, domain, name, and version without redefining its logic.

from typing import Type
from flytekit import ReferenceLaunchPlan

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