From af107c7158991b75dff0fcb16c37fa741888ee78 Mon Sep 17 00:00:00 2001 From: Jimmy Shen Date: Fri, 13 Dec 2024 17:19:02 -0800 Subject: [PATCH] added example for missing ref. --- examples/missing_reference.py | 36 +++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 examples/missing_reference.py diff --git a/examples/missing_reference.py b/examples/missing_reference.py new file mode 100644 index 00000000..d0f2aeb4 --- /dev/null +++ b/examples/missing_reference.py @@ -0,0 +1,36 @@ +"""A demonstration of sequential jobs with missing references.""" + +from jobflow import Flow, OnMissing, job, run_locally + + +@job +def func(fail: bool = False): + """Failable function.""" + if fail: + raise ValueError("An error occurred.") + return 1 + + +@job +def collect(job_outputs): + """Job that allows some parents to fail.""" + total = 0 + for jo in job_outputs: + if jo is None: + continue + total += jo + if total < 1: + raise ValueError("No enough finished parents.") + return total + + +job1, job2, job3 = func(), func(), func(fail=True) +job_outputs = [job1.output, job2.output, job3.output] +collect_job = collect(job_outputs) +collect_job.config.on_missing_references = OnMissing.NONE +flow = Flow([job1, job2, job3, collect_job]) + +# run the flow, you can +res = run_locally(flow) +n_finished = 2 +assert res[collect_job.uuid][1].output == n_finished