raise errors if run dag fails (#1251)

Author: eakmanrq

sqlmesh/cli/main.py

@@ -299,7 +299,9 @@ def plan(ctx: click.Context, environment: t.Optional[str] = None, **kwargs: t.An
 def run(ctx: click.Context, environment: t.Optional[str] = None, **kwargs: t.Any) -> None:
     """Evaluates the DAG of models using the built-in scheduler."""
     context = ctx.obj
-    context.run(environment, **kwargs)
+    success = context.run(environment, **kwargs)
+    if not success:
+        raise click.ClickException("Run DAG Failed. See output for details.")
 
 
 @cli.command("invalidate")

sqlmesh/core/context.py

@@ -394,7 +394,7 @@ def run(
         execution_time: t.Optional[TimeLike] = None,
         skip_janitor: bool = False,
         ignore_cron: bool = False,
-    ) -> None:
+    ) -> bool:
         """Run the entire dag through the scheduler.
 
         Args:
@@ -404,13 +404,16 @@ def run(
             execution_time: The date/time time reference to use for execution time. Defaults to now.
             skip_janitor: Whether to skip the janitor task.
             ignore_cron: Whether to ignore the model's cron schedule and run all available missing intervals.
+
+        Returns:
+            True if the run was successful, False otherwise.
         """
         environment = environment or c.PROD
         self.notification_target_manager.notify(
             NotificationEvent.RUN_START, environment=environment
         )
         try:
-            self.scheduler(environment=environment).run(
+            success = self.scheduler(environment=environment).run(
                 environment,
                 start=start,
                 end=end,
@@ -422,11 +425,21 @@ def run(
                 NotificationEvent.RUN_FAILURE, traceback.format_exc()
             )
             raise e
-        self.notification_target_manager.notify(NotificationEvent.RUN_END, environment=environment)
+        if success:
+            self.notification_target_manager.notify(
+                NotificationEvent.RUN_END, environment=environment
+            )
+        else:
+            self.notification_target_manager.notify(
+                NotificationEvent.RUN_FAILURE, environment=environment
+            )
+            return success
 
         if not skip_janitor and environment.lower() == c.PROD:
             self._run_janitor()
 
+        return success
+
     @t.overload
     def get_model(
         self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: Literal[True] = True

sqlmesh/magics.py

@@ -22,7 +22,7 @@
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.test import ModelTestMetadata, get_all_model_tests
 from sqlmesh.utils import sqlglot_dialects, yaml
-from sqlmesh.utils.errors import MagicError, MissingContextException
+from sqlmesh.utils.errors import MagicError, MissingContextException, SQLMeshError
 
 CONTEXT_VARIABLE_NAMES = [
     "context",
@@ -358,14 +358,16 @@ def run_dag(self, line: str) -> None:
         console = self._context.console
         self._context.console = get_console(display=self.display)
 
-        self._context.run(
+        success = self._context.run(
             args.environment,
             start=args.start,
             end=args.end,
             skip_janitor=args.skip_janitor,
             ignore_cron=args.ignore_cron,
         )
         self._context.console = console
+        if not success:
+            raise SQLMeshError("Error Running DAG. Check logs for details.")
 
     @magic_arguments()
     @argument("model", type=str, help="The model.")

tests/core/test_context.py

@@ -184,7 +184,7 @@ def test_diff(sushi_context: Context, mocker: MockerFixture):
     mock_console = mocker.Mock()
     sushi_context.console = mock_console
     yesterday = yesterday_ds()
-    sushi_context.run(start=yesterday, end=yesterday)
+    success = sushi_context.run(start=yesterday, end=yesterday)
 
     plan_evaluator = BuiltInPlanEvaluator(
         sushi_context.state_sync, sushi_context.snapshot_evaluator
@@ -198,6 +198,7 @@ def test_diff(sushi_context: Context, mocker: MockerFixture):
     sushi_context.upsert_model("sushi.customers", query=parse_one("select 1 as customer_id"))
     sushi_context.diff("test")
     assert mock_console.show_model_difference_summary.called
+    assert success
 
 
 def test_evaluate_limit():