mysterious-painter-66441
05/20/2025, 3:58 PMprint_num
, I want them to be unique based on the input.mysterious-painter-66441
05/20/2025, 3:58 PMfrom typing import Dict, List
from flytekit import LaunchPlan, WorkflowFailurePolicy, dynamic, task, workflow
@task
def retrieve_a_list() -> List[int]:
return [int(i) for i in range(20)]
@task
def print_num_task(num: int):
if num % 2 == 0:
raise Exception("num is odd")
print(num)
@workflow
def print_num(num: int):
print_num_task(num=num)
launch_plan = LaunchPlan.get_or_create(workflow=print_num)
@dynamic
def print_list(num_list: List[int]):
for num in num_list:
try:
launch_plan(num=num)
except:
print(f"Failed to print num: {num}")
@workflow(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def print_number_wf():
num_list = retrieve_a_list()
print_list(num_list=num_list)
if __name__ == "__main__":
print_number_wf()from typing import Dict, List
from flytekit import LaunchPlan, WorkflowFailurePolicy, dynamic, task, workflow
@task
def retrieve_a_list() -> List[int]:
return [int(i) for i in range(20)]
@task
def print_num_task(num: int):
if num % 2 == 0:
raise Exception("num is odd")
print(num)
@workflow
def print_num(num: int):
print_num_task(num=num)
launch_plan = LaunchPlan.get_or_create(workflow=print_num)
@dynamic
def print_list(num_list: List[int]):
for num in num_list:
try:
launch_plan(num=num)
except:
print(f"Failed to print num: {num}")
@workflow(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def print_number_wf():
num_list = retrieve_a_list()
print_list(num_list=num_list)
if __name__ == "__main__":
print_number_wf()
worried-airplane-87065
05/20/2025, 5:02 PMmysterious-painter-66441
05/20/2025, 7:16 PM