raise DagsterError for trying to re-execute a not failed run

dagster-io · Nov 1, 2023 · 5ad4a1e · 5ad4a1e
1 parent e2f9872
commit 5ad4a1e
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 10 deletions.
diff --git a/python_modules/dagster/dagster/_core/execution/api.py b/python_modules/dagster/dagster/_core/execution/api.py
@@ -327,10 +327,11 @@ def from_failure(run_id: str, instance: DagsterInstance) -> "ReexecutionOptions"
         from dagster._core.execution.plan.state import KnownExecutionState
 
         parent_run = check.not_none(instance.get_run_by_id(run_id))
-        check.invariant(
-            parent_run.status == DagsterRunStatus.FAILURE,
-            "Cannot reexecute from failure a run that is not failed",
-        )
+        if parent_run.status != DagsterRunStatus.FAILURE:
+            raise DagsterInvariantViolationError(
+                "Cannot reexecute from failure a run that is not failed"
+            )
+
         # Tried to thread through KnownExecutionState to execution plan creation, but little benefit.
         # It is recalculated later by the re-execution machinery.
         step_keys_to_execute, _ = KnownExecutionState.build_resume_retry_reexecution(

diff --git a/python_modules/dagster/dagster/_core/instance/__init__.py b/python_modules/dagster/dagster/_core/instance/__init__.py
@@ -1575,10 +1575,10 @@ def create_reexecuted_run(
         run_config = run_config if run_config is not None else parent_run.run_config
 
         if strategy == ReexecutionStrategy.FROM_FAILURE:
-            check.invariant(
-                parent_run.status == DagsterRunStatus.FAILURE,
-                "Cannot reexecute from failure a run that is not failed",
-            )
+            if parent_run.status != DagsterRunStatus.FAILURE:
+                raise DagsterInvariantViolationError(
+                    "Cannot reexecute from failure a run that is not failed",
+                )
 
             (
                 step_keys_to_execute,

diff --git a/python_modules/dagster/dagster_tests/execution_tests/test_execute_job.py b/python_modules/dagster/dagster_tests/execution_tests/test_execute_job.py
@@ -1,4 +1,3 @@
-import dagster._check as check
 import pytest
 from dagster import (
     AssetKey,
@@ -260,7 +259,7 @@ def test_reexecute_from_failure_successful_run(instance):
     ) as result:
         assert result.success
 
-    with pytest.raises(check.CheckError, match="run that is not failed"):
+    with pytest.raises(DagsterInvariantViolationError, match="run that is not failed"):
         ReexecutionOptions.from_failure(result.run_id, instance)