cuddly-jelly-27016
05/22/2025, 12:09 AM└── projecta
└── workflows
└── test_workflow.py (shown below)
Where test__workflow.py
from flytekit import LaunchPlan, dynamic, task, workflow
@task
def dummy_task(a: int) -> int:
return a
@workflow
def wrapper_workflow(a: int) -> int:
return dummy_task(a=a)
@dynamic
def dynamic_lp(a: int) -> int:
# In our real-world example, we are using LaunchPlan here rather than subworkflow to handle large fan out
lp = LaunchPlan.get_or_create(wrapper_workflow)
return lp(a=a)
@workflow
def workflow_test(a: int) -> int:
return dynamic_lp(a=a)
Project B
from flytekit import workflow
from flytekit.core.launch_plan import reference_launch_plan
def interface(a: int) -> int:
...
rlp = reference_launch_plan(
project="projecta",
domain="development",
name="projecta.workflows.test_workflow.workflow_test",
version="1",
)(interface)
@workflow
def workflow_calling_reference(a: int) -> int:
return rlp(a=a)
### Screenshots
No response
### Are you sure this issue hasn't been raised already?
• Yes
### Have you read the Code of Conduct?
• Yes
flyteorg/flytecuddly-jelly-27016
05/22/2025, 12:09 AM