Return completion status instead of using callback

Author: treysp

sqlmesh/core/plan/evaluator.py

@@ -37,7 +37,7 @@
     SnapshotInfoLike,
     SnapshotTableInfo,
 )
-from sqlmesh.core.scheduler import CompletionStatus
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.core.user import User
@@ -276,17 +276,20 @@ def _should_create(s: Snapshot) -> bool:
         completed = False
         progress_stopped = False
         try:
-            self.snapshot_evaluator.create(
+            completion_status = self.snapshot_evaluator.create(
                 snapshots_to_create,
                 snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
                 deployability_index=deployability_index,
                 on_start=lambda x: self.console.start_creation_progress(
                     x, plan.environment, self.default_catalog
                 ),
-                on_no_work=self.console.log_status_update,
                 on_complete=self.console.update_creation_progress,
             )
+            if completion_status.is_nothing_to_do:
+                self.console.log_status_update(
+                    "\n[green]SKIP: No physical layer updates to perform[/green]\n"
+                )
             completed = True
         except NodeExecutionFailedError as ex:
             self.console.stop_creation_progress(success=False)

sqlmesh/core/scheduler.py

@@ -1,5 +1,4 @@
 from __future__ import annotations
-from enum import Enum
 import logging
 import typing as t
 from sqlglot import exp
@@ -31,6 +30,7 @@
     parent_snapshots_by_name,
 )
 from sqlmesh.core.state_sync import StateSync
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.utils.concurrency import concurrent_apply_to_dag, NodeExecutionFailedError
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -48,24 +48,6 @@
 SchedulingUnit = t.Tuple[str, t.Tuple[Interval, int]]
 
 
-class CompletionStatus(Enum):
-    SUCCESS = "success"
-    FAILURE = "failure"
-    NOTHING_TO_DO = "nothing_to_do"
-
-    @property
-    def is_success(self) -> bool:
-        return self == CompletionStatus.SUCCESS
-
-    @property
-    def is_failure(self) -> bool:
-        return self == CompletionStatus.FAILURE
-
-    @property
-    def is_nothing_to_do(self) -> bool:
-        return self == CompletionStatus.NOTHING_TO_DO
-
-
 class Scheduler:
     """Schedules and manages the evaluation of snapshots.
 

sqlmesh/core/snapshot/evaluator.py

@@ -51,7 +51,7 @@
     ViewKind,
     CustomKind,
 )
-
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.core.schema_diff import has_drop_alteration, get_dropped_column_names
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
@@ -279,20 +279,21 @@ def create(
         snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
         on_start: t.Optional[t.Callable] = None,
-        on_no_work: t.Optional[t.Callable] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
-    ) -> None:
+    ) -> CompletionStatus:
         """Creates a physical snapshot schema and table for the given collection of snapshots.
 
         Args:
             target_snapshots: Target snapshots.
             snapshots: Mapping of snapshot ID to snapshot.
             deployability_index: Determines snapshots that are deployable in the context of this creation.
             on_start: A callback to initialize the snapshot creation progress bar.
-            on_no_work: A callback to call when no snapshots are to be created.
             on_complete: A callback to call on each successfully created snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
+
+        Returns:
+            CompletionStatus: The status of the creation operation (success, failure, nothing to do).
         """
         snapshots_with_table_names = defaultdict(set)
         tables_by_schema = defaultdict(set)
@@ -350,9 +351,7 @@ def _get_data_objects(schema: exp.Table, gateway: t.Optional[str] = None) -> t.S
                 target_deployability_flags[snapshot.name].sort()
 
         if not snapshots_to_create:
-            if on_no_work:
-                on_no_work("\n[green]SKIP: No physical layer updates to perform[/green]\n")
-            return
+            return CompletionStatus.NOTHING_TO_DO
         if on_start:
             on_start(len(snapshots_to_create))
         self._create_schemas(tables_by_schema, gateway_by_schema)
@@ -364,6 +363,7 @@ def _get_data_objects(schema: exp.Table, gateway: t.Optional[str] = None) -> t.S
             on_complete=on_complete,
             allow_destructive_snapshots=allow_destructive_snapshots,
         )
+        return CompletionStatus.SUCCESS
 
     def _create_snapshots(
         self,

sqlmesh/utils/__init__.py

@@ -16,7 +16,7 @@
 from collections import defaultdict
 from contextlib import contextmanager
 from copy import deepcopy
-from enum import IntEnum
+from enum import IntEnum, Enum
 from functools import lru_cache, reduce, wraps
 from pathlib import Path
 
@@ -347,3 +347,21 @@ class Verbosity(IntEnum):
     DEFAULT = 0
     VERBOSE = 1
     VERY_VERBOSE = 2
+
+
+class CompletionStatus(Enum):
+    SUCCESS = "success"
+    FAILURE = "failure"
+    NOTHING_TO_DO = "nothing_to_do"
+
+    @property
+    def is_success(self) -> bool:
+        return self == CompletionStatus.SUCCESS
+
+    @property
+    def is_failure(self) -> bool:
+        return self == CompletionStatus.FAILURE
+
+    @property
+    def is_nothing_to_do(self) -> bool:
+        return self == CompletionStatus.NOTHING_TO_DO