Feat: Interrupt Flag (#251)

* Introduced  `--fail-fast` flag to cause sayn to skip the remaining execution if a Task in the DAG fails.

sayn/cli.py

@@ -24,6 +24,7 @@ def __init__(
         full_load=False,
         start_dt=None,
         end_dt=None,
+        fail_fast=False,
     ):
         super().__init__()
 
@@ -68,6 +69,9 @@ def __init__(
         if upstream_prod is not None:
             self.run_arguments.upstream_prod = upstream_prod
 
+        if fail_fast is not None:
+            self.run_arguments.fail_fast = fail_fast
+
         self.start_app()
 
 
@@ -117,6 +121,13 @@ def parser_process(value, state):
     "--debug", "-d", is_flag=True, default=False, help="Include debug messages"
 )
 
+click_fail_fast = click.option(
+    "--fail-fast",
+    is_flag=True,
+    default=False,
+    help="Interrupt remaining task execution on first failure.",
+)
+
 
 def click_filter(func):
     func = click.option(
@@ -168,6 +179,7 @@ def click_incremental(func):
 
 def click_run_options(func):
     func = click_debug(func)
+    func = click_fail_fast(func)
     func = click.option("--profile", "-p", help="Profile from settings to use")(func)
     func = click_incremental(func)
     func = click_filter(func)
@@ -190,7 +202,17 @@ def init(sayn_project_name):
 
 @cli.command(help="Compile sql tasks.")
 @click_run_options
-def compile(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end_dt):
+def compile(
+    debug,
+    tasks,
+    exclude,
+    upstream_prod,
+    profile,
+    full_load,
+    start_dt,
+    end_dt,
+    fail_fast,
+):
 
     tasks = [i for t in tasks for i in t.strip().split(" ")]
     exclude = [i for t in exclude for i in t.strip().split(" ")]
@@ -204,6 +226,7 @@ def compile(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt,
         full_load,
         start_dt,
         end_dt,
+        fail_fast,
     )
 
     app.compile()
@@ -215,7 +238,17 @@ def compile(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt,
 
 @cli.command(help="Run SAYN tasks.")
 @click_run_options
-def run(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end_dt):
+def run(
+    debug,
+    tasks,
+    exclude,
+    upstream_prod,
+    profile,
+    full_load,
+    start_dt,
+    end_dt,
+    fail_fast,
+):
 
     tasks = [i for t in tasks for i in t.strip().split(" ")]
     exclude = [i for t in exclude for i in t.strip().split(" ")]
@@ -229,6 +262,7 @@ def run(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end_
         full_load,
         start_dt,
         end_dt,
+        fail_fast,
     )
 
     app.run()
@@ -240,7 +274,17 @@ def run(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end_
 
 @cli.command(help="Test SAYN tasks.")
 @click_run_options
-def test(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end_dt):
+def test(
+    debug,
+    tasks,
+    exclude,
+    upstream_prod,
+    profile,
+    full_load,
+    start_dt,
+    end_dt,
+    fail_fast,
+):
 
     tasks = [i for t in tasks for i in t.strip().split(" ")]
     exclude = [i for t in exclude for i in t.strip().split(" ")]
@@ -254,6 +298,7 @@ def test(debug, tasks, exclude, upstream_prod, profile, full_load, start_dt, end
         full_load,
         start_dt,
         end_dt,
+        fail_fast,
     )
 
     app.test()

sayn/core/app.py

@@ -68,6 +68,7 @@ class Folders:
     command: Command = Command.UNDEFINED
     upstream_prod: bool = False
     is_prod: bool = False
+    fail_fast: bool = False
 
     include: Set[str]
     exclude: Set[str]
@@ -609,11 +610,15 @@ def execute_dag(self):
         self.tracker.start_stage(
             self.run_arguments.command.value, tasks=list(tasks_in_query.keys())
         )
+        interrupt_flag = False
 
         for task in self.tasks.values():
             if not task.in_query:
                 continue
 
+            if interrupt_flag:
+                task.fail_fast = True
+
             # We force the run/compile so that the skipped status can be calculated,
             # but we only report if the task is in the query
             # if task.in_query:
@@ -634,6 +639,9 @@ def execute_dag(self):
                     "finish_stage", duration=datetime.now() - start_ts, result=result
                 )
 
+            if self.run_arguments.fail_fast and result.is_err:
+                interrupt_flag = True
+
         self.tracker.finish_current_stage(
             tasks={k: v.status for k, v in tasks_in_query.items()},
             test=True if self.run_arguments.command == Command.TEST else False,
@@ -643,7 +651,19 @@ def execute_dag(self):
 
     def finish_app(self, error=None):
         duration = datetime.now() - self.app_start_ts
-        if error is not None:
+        if self.run_arguments.fail_fast and error is not None:
+            self.tracker.report_event(
+                event="finish_stage",
+                duration=duration,
+                tasks={
+                    k: v.status
+                    if v.status in (TaskStatus.SUCCEEDED, TaskStatus.FAILED)
+                    else TaskStatus.SKIPPED
+                    for k, v in self.tasks.items()
+                },
+            )
+            sys.exit(-1)
+        elif error is not None:
             self.tracker.report_event(
                 event="finish_app",
                 duration=duration,

sayn/logging/fancy_logger.py

@@ -52,6 +52,9 @@ def task_stage_finish(self, stage, duration, result):
             if result.error.code == "parent_errors":
                 self.spinner.text_color = "yellow"
                 self.spinner.warn()
+            if result.error.code == "interrupted":
+                self.spinner.text_color = "yellow"
+                self.spinner.warn()
             elif (
                 result.error.code == "setup_error"
                 and result.error.details["status"].value == "skipped"

sayn/logging/log_formatter.py

@@ -286,6 +286,10 @@ def error_result(self, duration, error):  # noqa: C901
                 f"Skipping due to ancestors errors: {parents} ({duration})"
             )
 
+        elif error.code == "interrupted":
+            level = "warning"
+            message = self.warn(f"Skipping due to Interrupt signal ({duration})")
+
         elif error.code == "setup_error":
             if error.details["status"].value == "skipped":
                 level = "warning"

sayn/tasks/task_wrapper.py

@@ -88,6 +88,7 @@ def __init__(
         self.used_connections = set()
         self.tracker = tracker
         self.runner = None
+        self.fail_fast = False
 
         self.name = name
         self.group = group
@@ -266,6 +267,10 @@ def set_parameters(
         self.compiler.update_globals(**task_parameters)
 
     def check_skip(self):
+        if self.fail_fast:
+            self.status = TaskStatus.SKIPPED
+            return Err("task", "interrupted")
+
         if self.run_arguments["command"] != "test":
             failed_parents = {
                 p.name: p.status