Merge remote-tracking branch 'materialsproject/main' into uuid_patch

src/jobflow/managers/local.py

@@ -21,6 +21,7 @@ def run_locally(
     root_dir: str | Path | None = None,
     ensure_success: bool = False,
     allow_external_references: bool = False,
+    raise_immediately: bool = False,
 ) -> dict[str, dict[int, jobflow.Response]]:
     """
     Run a :obj:`Job` or :obj:`Flow` locally.
@@ -46,6 +47,10 @@ def run_locally(
     allow_external_references : bool
         If False all the references to other outputs should be from other Jobs
         of the Flow.
+    raise_immediately : bool
+        If True, raise an exception immediately if a job fails. If False, continue
+        running the flow and only raise an exception at the end if the flow did not
+        finish running successfully.
 
     Returns
     -------
@@ -102,14 +107,19 @@ def _run_job(job: jobflow.Job, parents):
             errored.add(job.uuid)
             return None, False
 
-        try:
+        if raise_immediately:
             response = job.run(store=store)
-        except Exception:
-            import traceback
-
-            logger.info(f"{job.name} failed with exception:\n{traceback.format_exc()}")
-            errored.add(job.uuid)
-            return None, False
+        else:
+            try:
+                response = job.run(store=store)
+            except Exception:
+                import traceback
+
+                logger.info(
+                    f"{job.name} failed with exception:\n{traceback.format_exc()}"
+                )
+                errored.add(job.uuid)
+                return None, False
 
         responses[job.uuid][job.index] = response
 

tests/managers/test_local.py

@@ -331,6 +331,9 @@ def test_error_flow(memory_jobstore, clean_dir, error_flow, capsys):
     with pytest.raises(RuntimeError):
         run_locally(flow, store=memory_jobstore, ensure_success=True)
 
+    with pytest.raises(ValueError, match="errored"):
+        run_locally(flow, store=memory_jobstore, raise_immediately=True)
+
 
 def test_ensure_success_with_replace(memory_jobstore, error_replace_flow, capsys):
     from jobflow import run_locally