Execute launchplans declared locally and automatically adjust input params based on fixed and default values

[kumare3]

Today, you could execute local launchplans, but the default and fixed values were not respected. This fix solve this problem.

Consider this example

@task
def two_inputs(x: int, y: str) -> str:
    return f"{x},{y}"

@workflow
def two_inputs_wf(x: int, y: str) -> str:
    return two_inputs(x, y)

lp_fixed_y_default_x = LaunchPlan.get_or_create(
    workflow=two_inputs_wf,
    name="fixed-default-inputs",
    fixed_inputs={"y": "hello"},
    default_inputs={"x": 1}
)

lp_fixed_y = LaunchPlan.get_or_create(
    workflow=two_inputs_wf,
    name="fixed-y",
    fixed_inputs={"y": "hello"},
)

lp_fixed_x = LaunchPlan.get_or_create(
    workflow=two_inputs_wf,
    name="fixed-x",
    fixed_inputs={"x": 1},
)

lp_fixed_all = LaunchPlan.get_or_create(
    workflow=two_inputs_wf,
    name="fixed-all",
    fixed_inputs={"x": 1, "y": "test"},
)

lp_default_x = LaunchPlan.get_or_create(
    name="default-inputs",
    workflow=two_inputs_wf,
    default_inputs={"x": 1}
)

lp_simple = LaunchPlan.get_or_create(
    workflow=two_inputs_wf,
    name="no-fixed-default",
)

Can be invoked correctly

pyflyte run test.py lp_simple --x 10 --y hello

(flytekit) ➜  testing git:(local-lp-fixed-inputs) pyflyte run test.py lp_fixed_y_default_x                                                                                                                                      
Running Execution on local.
1,hello

TODO I am observing a bug in admin for registeration where version is ignored

Summary by Bito

This PR improves LaunchPlan local execution by implementing proper handling of fixed and default input values. The changes focus on the WorkflowCommand class to ensure fixed inputs cannot be overridden while maintaining optional default inputs. The implementation includes comprehensive test coverage for various input combinations.

Unit tests added: True

Estimated effort to review (1-5, lower is better): 2

[flyte-bot]

Code Review Agent Run #58cf8a

Actionable Suggestions - 2

• tests/flytekit/unit/cli/pyflyte/test_run.py - 1

  • Consider using pathlib for file operations · Line 324-332
• flytekit/clis/sdk_in_container/run.py - 1

  • Consider extracting launch plan input logic · Line 1055-1069

Additional Suggestions - 1

• tests/flytekit/unit/cli/pyflyte/test_run.py - 1

  • Consider adjusting parameter indentation for readability · Line 334-335

Review Details

• Files reviewed - 4 · Commit Range: 1b56221..c0bcb27

  • flytekit/clis/sdk_in_container/run.py
  • tests/flytekit/unit/cli/pyflyte/test_run.py
  • tests/flytekit/unit/cli/pyflyte/test_run_lps.py
  • tests/flytekit/unit/cli/pyflyte/workflow.py
• Files skipped - 1

  • .gitignore - Reason: Filter setting
• Tools

  • Whispers (Secret Scanner) - ✔︎ Successful
  • Detect-secrets (Secret Scanner) - ✔︎ Successful
  • MyPy (Static Code Analysis) - ✔︎ Successful
  • Astral Ruff (Static Code Analysis) - ✔︎ Successful

---

AI Code Review powered by

[flyte-bot]

Changelist by Bito

This pull request implements the following key changes.

Key Change	Files Impacted
Feature Improvement - Launch Plan Input Handling Enhancement	- run.py - Added support for handling fixed and default inputs in LaunchPlan execution - test_run_lps.py - Added comprehensive tests for LaunchPlan execution with fixed and default inputs - test_run.py - Updated imports and code formatting improvements - workflow.py - Added LaunchPlan import for testing

[flyte-bot]

Consider using pathlib for file operations

Consider simplifying the nested function calls by using pathlib.Path for file operations. This could make the code more readable and maintainable.

Code suggestion

Check the AI-generated fix before applying

 -                str(
 -                    json.load(
 -                        open(
 -                            os.path.join(
 -                                os.path.dirname(os.path.realpath(__file__)),
 -                                "my_wf_input.json",
 -                            ),
 -                            "r",
 -                        )
 -                    )
 -                ),
 +                str(json.loads(pathlib.Path(__file__).parent.joinpath("my_wf_input.json").read_text())),

Code Review Run #58cf8a

---

Is this a valid issue, or was it incorrectly flagged by the Agent?

• it was incorrectly flagged

[flyte-bot]

Consider extracting launch plan input logic

The launch plan input handling logic could be extracted into a separate helper method to improve code readability and maintainability. Consider moving the logic for handling fixed and default inputs into a dedicated function.

Code suggestion

Check the AI-generated fix before applying

Suggested change

	final_inputs_with_defaults = loaded_entity.python_interface.inputs_with_defaults
	if isinstance(loaded_entity, LaunchPlan):
	# For LaunchPlans it is essential to handle fixed inputs and default inputs in a special way
	# Fixed inputs are inputs that are always passed to the launch plan and cannot be overridden
	# Default inputs are inputs that are optional and have a default value
	# The final inputs to the launch plan are a combination of the fixed inputs and the default inputs
	all_inputs = loaded_entity.python_interface.inputs_with_defaults
	default_inputs = loaded_entity.saved_inputs
	pmap = loaded_entity.parameters
	final_inputs_with_defaults = {}
	for name, _ in pmap.parameters.items():
	_type, v = all_inputs[name]
	if name in default_inputs:
	v = default_inputs[name]
	final_inputs_with_defaults[name] = _type, v
	final_inputs_with_defaults = self._get_launch_plan_inputs(loaded_entity) if isinstance(loaded_entity, LaunchPlan) else loaded_entity.python_interface.inputs_with_defaults
	def _get_launch_plan_inputs(self, launch_plan):
	# For LaunchPlans it is essential to handle fixed inputs and default inputs in a special way
	# Fixed inputs are inputs that are always passed to the launch plan and cannot be overridden
	# Default inputs are inputs that are optional and have a default value
	# The final inputs to the launch plan are a combination of the fixed inputs and the default inputs
	all_inputs = launch_plan.python_interface.inputs_with_defaults
	default_inputs = launch_plan.saved_inputs
	pmap = launch_plan.parameters
	final_inputs = {}
	for name, _ in pmap.parameters.items():
	_type, v = all_inputs[name]
	if name in default_inputs:
	v = default_inputs[name]
	final_inputs[name] = _type, v
	return final_inputs

Code Review Run #58cf8a

---

Is this a valid issue, or was it incorrectly flagged by the Agent?

• it was incorrectly flagged

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 91.08%. Comparing base (9d34416) to head (c0bcb27).

Additional details and impacted files

@@             Coverage Diff             @@
##           master    #3115       +/-   ##
===========================================
+ Coverage   78.20%   91.08%   +12.87%     
===========================================
  Files         292      114      -178     
  Lines       25401     5785    -19616     
  Branches     2779        0     -2779     
===========================================
- Hits        19864     5269    -14595     
+ Misses       4726      516     -4210     
+ Partials      811        0      -811     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[kumare3]

Cc @fiedlerNr9 @eapolinario
If you run the second one admin rejects with missin version (cryptic bug)