Dominick Olivito
03/30/2022, 12:27 AMprint("hello world 1")
if not skip_task2:
print("hello world 2")
if not skip_task3:
print("hello world 3")
print("hello world 4")
here's an attempt in Prefect:
@task()
def print_task(value: str) -> str:
logger = prefect.context.get("logger")
<http://logger.info|logger.info>(value)
return value
with Flow("case_skip_flow") as flow:
skip_task2 = Parameter("skip_task2", True)
skip_task3 = Parameter("skip_task3", True)
task1 = print_task("hello world 1")
with case(skip_task2, False):
task2 = print_task("hello world 2", upstream_tasks=[task1])
with case(skip_task3, False):
task3 = print_task(
"hello world 3", upstream_tasks=[task1, task2], task_args={"skip_on_upstream_skip": False}
)
task4 = print_task("hello world 4", upstream_tasks=[task3], task_args={"skip_on_upstream_skip": False})
i'm getting into trouble with the upstream skip status for task3
.
• if I set skip_on_upstream_skip
to True
, then when task2
is skipped, so is task3
regardless of the value for skip_task3
• if I set skip_on_upstream_skip
to False
, then task3
runs regardless of the value for skip_task3
how can i implement a pattern like this?Kevin Kho
03/30/2022, 1:31 AMDominick Olivito
03/30/2022, 1:42 PM@task()
def dummy_task() -> None:
pass
with Flow("case_skip_flow") as flow:
skip_task2 = Parameter("skip_task2", True)
skip_task3 = Parameter("skip_task3", False)
task1 = print_task("hello world 1")
with case(skip_task2, False):
task2 = print_task("hello world 2", upstream_tasks=[task1])
dummy_result = dummy_task(upstream_tasks=[task2], task_args={"skip_on_upstream_skip": False})
with case(skip_task3, False):
task3 = print_task(
"hello world 3", upstream_tasks=[dummy_result]
)
task4 = print_task("hello world 4", upstream_tasks=[task3])
Kevin Kho
03/30/2022, 1:43 PM.wait()
and use the native Python if