Launch plans, schedules, and fixed inputs

Overview

A launch plan packages a workflow together with a set of execution parameters — default inputs, fixed inputs, a schedule, notifications, labels, annotations, security settings, and more. Every workflow registered in Flyte gets a default launch plan automatically; named launch plans let you override any of those properties.

The LaunchPlan class lives in flytekit.core.launch_plan and is the entry point for creating, caching, and invoking parameterized workflow executions.

Creating Launch Plans

There are two factory methods on the LaunchPlan class, both in launch_plan.py.

Default launch plan (no name)

from flytekit import LaunchPlan, workflow

@workflow
def my_wf(a: int, b: int = 5) -> int:
    ...

default_lp = LaunchPlan.get_or_create(workflow=my_wf)

When you call get_or_create without a name, flytekit produces a default launch plan. A default plan picks up whatever default values are already defined in the workflow function signature, uses the default auth information supplied during serialization, and carries no schedules, notifications, or additional attributes. Trying to pass schedules, fixed inputs, or any other property without a name raises a ValueError:

# This raises ValueError
LaunchPlan.get_or_create(
    workflow=my_wf,
    schedule=some_schedule,  # not allowed on a default plan
)

The guard is in get_or_create at line 231 of launch_plan.py. It checks every optional parameter — if any is set while name is None, the ValueError is raised with the message: "Only named launchplans can be created that have other properties."

Named launch plan

If you supply a name, the launch plan becomes a full-featured object:

LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_custom_lp",
    default_inputs={"a": 10},
    fixed_inputs={"b": 20},
)

Named plans are cached so that calling get_or_create again with the same name returns the same object — but only if all attributes match. If you try to create two plans with the same name but different attribute values, flytekit raises an AssertionError.

The create() classmethod

LaunchPlan.create() is an alternative that always creates a fresh plan (though it still checks the cache for duplicate names). Its signature mirrors get_or_create but requires a name as the first positional argument:

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

lp = LaunchPlan.create(
    "schedule_test",
    quadruple,
    schedule=FixedRate(duration=timedelta(hours=12), kickoff_time_input_arg="kickoff_input"),
)

This is used internally by get_or_create but is also available if you want to bypass the caching logic and explicitly construct a plan. Duplicate names still raise AssertionError.

Default and Fixed Inputs

The distinction between default_inputs and fixed_inputs is fundamental to how launch plans control workflow execution.

Default inputs (overridable)

Default inputs behave like function keyword-argument defaults. When you create the launch plan, you provide values that will be used unless the caller supplies a different value at invocation time. They are merged into the workflow's parameter map.

Internally, create() merges defaults from two sources (line 121 of launch_plan.py): the workflow function signature's own Python defaults, and the default_inputs argument passed to create() or get_or_create(). The latter takes precedence.

Fixed inputs (frozen)

Fixed inputs are values that cannot be overridden at call time. They are converted to LiteralMap literals and stored separately on the launch plan. The constructor then removes them from the parameter map (line 338 of launch_plan.py), stripping out any key that appears in fixed_inputs.literals. This means any attempt to pass the same input key at execution time has no effect — the fixed value always wins.

A typical example:

LaunchPlan.get_or_create(
    workflow=wf,
    name="your_lp_name_1",
    default_inputs={"a": 3},
    fixed_inputs={"c": "4"},
)

Here "a" can be overridden by the caller; "c" is locked to "4".

What happens when you call a launch plan

When you invoke a launch plan, saved_inputs — a dict that merges both default and fixed inputs — is combined with any keyword arguments you pass. See __call__ in launch_plan.py (line 465):

def __call__(self, *args, **kwargs):
    if len(args) > 0:
        raise AssertionError("Only Keyword Arguments are supported for launch plan executions")

    ctx = FlyteContext.current_context()
    if ctx.compilation_state is not None:
        inputs = self.saved_inputs
        inputs.update(kwargs)
        return create_and_link_node(ctx, entity=self, **inputs)
    else:
        inputs = self.saved_inputs
        inputs.update(kwargs)
        return self.workflow(*args, **inputs)

The merge order is: saved inputs (defaults + fixed) first, then call-site kwargs overwriting where applicable. For fixed inputs, since they are already removed from the parameter map, the caller cannot even pass them — the merged dict simply doesn't include those keys.

Workflow signature defaults are inherited

When you create a default launch plan, get_default_launch_plan extracts defaults from the workflow's python_interface.inputs_with_defaults:

default_inputs = {
    name: default for name, (type, default) in workflow.python_interface.inputs_with_defaults.items()
}
lp._saved_inputs = default_inputs

So if your workflow has a keyword argument with a default, the default launch plan already knows about it. This is why calling lp(a=8) inside another workflow correctly resolves to the sum 8 + 5 when b=5 is the workflow default:

@task
def t1(a: int, b: int) -> int:
    return a + b

@workflow
def my_sub_wf(a: int, b: int = 5) -> int:
    return t1(a=a, b=b)

lp = LaunchPlan.get_or_create(my_sub_wf)

@workflow
def my_wf(a: int) -> int:
    return lp(a=a)

assert my_wf(a=8) == 13  # 8 + 5

Scheduling and Triggers

Launch plans can be configured to run automatically on a schedule. Flytekit provides two schedule implementations in flytekit.core.schedule — CronSchedule and FixedRate — plus a newer trigger parameter that wraps either one.

CronSchedule

CronSchedule (defined in schedule.py, line 21) accepts a schedule parameter that can be either a cron alias or a 5-field cron expression. The accepted aliases are:

hourly, hours, @hourly, daily, days, @daily, weekly, weeks, @weekly, monthly, months, @monthly, annually, @annually, yearly, years, @yearly

Validation uses the croniter library:

@staticmethod
def _validate_schedule(schedule: str):
    if schedule.lower() not in CronSchedule._VALID_CRON_ALIASES:
        try:
            croniter.croniter(schedule)
        except Exception:
            raise ValueError(
                "Schedule is invalid. It must be set to either a cron alias or valid cron expression."
            )

A 5-field cron expression like "*/1 * * * *" runs every minute. The deprecated cron_expression parameter (for 6-field AWS-style expressions) raises AssertionError if used.

You can optionally pass an ISO 8601 offset string (validated against the regex ([-+]?)P([-+0-9YMWD]+)?(T([-+0-9HMS.,]+)?)?) and a kickoff_time_input_arg that injects the scheduled trigger time into your workflow:

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

CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

FixedRate

FixedRate (line 158) takes a datetime.timedelta and automatically translates it to the most appropriate unit — days, hours, or minutes:

from datetime import timedelta

FixedRate(duration=timedelta(hours=10), kickoff_time_input_arg="abc")
# rate.unit = HOUR, rate.value = 10

FixedRate(duration=timedelta(hours=24))
# rate.unit = DAY, rate.value = 1  (24h is normalized to 1 day)

FixedRate(duration=timedelta(minutes=30))
# rate.unit = MINUTE, rate.value = 30

The translation logic in _translate_duration (line 178) checks divisors in order: days, then hours, then minutes. Sub-minute granularity raises AssertionError:

if duration.microseconds != 0 or duration.seconds % 60 != 0:
    raise AssertionError(
        "Granularity of less than a minute is not supported for FixedRate schedules."
    )

Attaching a schedule to a launch plan

You can pass a CronSchedule or FixedRate directly to the schedule parameter:

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

LaunchPlan.create(
    "schedule_test",
    quadruple,
    schedule=FixedRate(duration=timedelta(hours=12), kickoff_time_input_arg="kickoff_input"),
)

This stores the schedule on the launch plan's schedule property. Round-tripping through protobuf is supported via to_flyte_idl() and from_flyte_idl().

The newer trigger parameter

OnSchedule (line 209) is a lightweight wrapper that implements the LaunchPlanTriggerBase protocol:

class LaunchPlanTriggerBase(Protocol):
    def to_flyte_idl(self, *args, **kwargs) -> google_message.Message: ...

class OnSchedule(LaunchPlanTriggerBase):
    def __init__(self, schedule: Union[CronSchedule, FixedRate]):
        self._schedule = schedule

    def to_flyte_idl(self) -> schedule_pb2.Schedule:
        return self._schedule.to_flyte_idl()

Pass it as the trigger keyword argument instead of schedule:

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

lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="triggered_lp",
    trigger=OnSchedule(CronSchedule(schedule="@daily")),
)

The trigger parameter is marked as [alpha] in the source — use it with that understanding.

Notifications alongside schedules

A launch plan's notifications parameter takes a list of Notification objects, each tied to a workflow execution phase:

from flytekit.models.common import Notification
from flytekit.models.core.execution import WorkflowExecutionPhase

email_notif = Notification(
    phases=[WorkflowExecutionPhase.SUCCEEDED],
    recipients_email=["my-team@email.com"],
)

LaunchPlan.get_or_create(
    workflow=wf,
    name="your_lp_name_2",
    schedule=CronSchedule(schedule="*/1 * * * *"),
    notifications=[email_notif],
)

Calling Launch Plans in Workflows

A launch plan is a first-class callable entity inside a workflow. When you call lp(**kwargs), flytekit merges the launch plan's saved inputs (defaults + fixed) with the call-site kwargs and delegates to the underlying workflow.

@workflow
def parent_wf(x: int) -> int:
    return lp(a=x)  # saved_inputs provide other arguments

During compilation (inside @workflow), the call goes through create_and_link_node, which wires the launch plan as a sub-node in the DAG. During local execution (outside a workflow), it delegates directly to the workflow function with the merged inputs.

Only keyword arguments are supported — passing positional arguments raises AssertionError.

Other Attributes

Launch plans support a wide range of optional metadata. Each is set by passing the appropriate model object:

Parameter	Model	Purpose
labels	Labels({"key": "value"})	Kubernetes-style labels on executions
annotations	Annotations({"key": "value"})	Kubernetes-style annotations on executions
auth_role	AuthRole(assumable_iam_role="my:role")	IAM role or K8s service account (deprecated)
security_context	SecurityContext(...)	Replacement for auth_role
raw_output_data_config	RawOutputDataConfig("s3://bucket/path")	Offloaded data location
max_parallelism	int	Max parallel task nodes in the workflow
overwrite_cache	bool	Always overwrite cached task outputs
auto_activate	bool (default False)	Activate on registration

Example from the documentation tests:

from flytekit.models.common import Annotations, AuthRole, Labels, RawOutputDataConfig

labels_model = Labels({"label": "foo"})
annotations_model = Annotations({"annotate": "bar"})

LaunchPlan.get_or_create(
    workflow=wf,
    name="your_lp_name_4",
    auth_role=AuthRole(assumable_iam_role="my:iam:role"),
    labels=labels_model,
    annotations=annotations_model,
)

Note that auth_role is deprecated. If you pass both auth_role and security_context, create() raises ValueError.

Workflow default options propagate

When you create a default launch plan via get_or_create, flytekit reads the workflow's default_options to populate the plan's labels and annotations automatically:

if workflow.default_options is not None:
    default_labels = workflow.default_options.labels
    default_annotations = workflow.default_options.annotations

Imperative Workflow API

If you prefer to build workflows imperatively rather than with decorators, WorkflowBase.add_launch_plan() in workflow.py adds a launch plan as a node:

wf = WorkflowBase(...)
wf.add_launch_plan(lp, a=input_a, b=input_b)

This is equivalent to the decorator-style lp(a=input_a, b=input_b) call inside a @workflow function. The same method exists for tasks (add_task) and sub-workflows (add_subwf).

Array Mapping over Launch Plans

The ArrayNode class in array_node.py accepts a LaunchPlan as its target for map-style execution. When the target is a launch plan (not a ReferenceLaunchPlan), fixed inputs are automatically detected and excluded from the mapped interface:

if isinstance(target, (LaunchPlan, FlyteLaunchPlan)) and not isinstance(target, ReferenceLaunchPlan):
    self._excluded_inputs = set(target.fixed_inputs.literals)

This ensures that fixed inputs like a database connection string or a configuration parameter are not treated as per-element mapping inputs. The remaining inputs become the list-shaped interface of the array node.

Reference Launch Plans

A ReferenceLaunchPlan (launch_plan.py, line 482) is a pointer to a launch plan that already exists on a Flyte installation. It does not make a network call; the user provides the expected interface via type-annotated function arguments.

Use the reference_launch_plan() decorator:

from flytekit import reference_launch_plan

@reference_launch_plan(project="my_project", domain="production", name="existing_lp", version="v1")
def existing_lp(a: int, b: str) -> bool:
    """The signature must match the remote launch plan's interface."""
    ...

Or use the generic get_reference_entity() from flytekit.core.reference for programmatic usage:

from flytekit.models.core.identifier import ResourceType
from flytekit.core.reference import get_reference_entity

ref_lp = get_reference_entity(
    ResourceType.LAUNCH_PLAN,
    project="my_project",
    domain="dev",
    name="my.launch.plan",
    version="abc123",
    inputs={"a": int, "b": str},
    outputs={},
)

A ReferenceLaunchPlan cannot be executed locally — calling it raises NotImplementedError. Its purpose is compilation-time validation; errors surface only when the remote interface differs from what was declared.

Dynamic Task Dependencies

When you call a launch plan from inside a dynamic task, flytekit cannot automatically discover that dependency before runtime. You must list the launch plan in the node_dependency_hints parameter so that registration includes it:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    return [launchplan0] * 10

Without the hint, the launch plan may not be registered on the Flyte admin server, and the dynamic execution would fail at runtime.

Serialization and Registration

Serializing a launch plan to protobuf (via get_serializable()) requires SerializationSettings with project, domain, version, and image_config. The launch plan's fixed inputs are already stored as a LiteralMap ready for protobuf output. The CronSchedule and FixedRate objects implement to_flyte_idl() for their protobuf round-trip.

The auto_activate parameter controls whether the launch plan is automatically activated upon registration. By default it is False.

Gotchas and Best Practices

• No extra attributes on default plans. If you call get_or_create without a name, you cannot pass schedules, fixed inputs, or any other parameter — doing so raises ValueError.
• Names must be unique per workflow. Creating two named launch plans for the same workflow with different attribute values raises AssertionError.
• Fixed inputs are immutable. Once set, they are removed from the parameter map and cannot be overridden at call time.
• cron_expression is deprecated. Pass the schedule parameter instead; using cron_expression raises AssertionError.
• Sub-minute FixedRate raises an error. Durations with microseconds or seconds not divisible by 60 raise AssertionError.
• Only keyword arguments in __call__. Positional arguments raise AssertionError.
• auth_role is deprecated. Pass security_context instead. Using both raises ValueError.
• Cached plans are shared. get_or_create returns the same object for the same name. Mutating the returned object affects all references.
• saved_inputs returns a copy. Callers can mutate the returned dict without affecting the launch plan's internal state.