Add live phase-checklist progress for snow dcm deploy and plan

src/snowflake/cli/_plugins/dcm/commands.py

@@ -23,6 +23,7 @@
 )
 from snowflake.cli._plugins.dcm.manager import DCMProjectManager
 from snowflake.cli._plugins.dcm.models import DCMManifest, DCMTarget, TargetContext
+from snowflake.cli._plugins.dcm.progress import DeployProgressTracker
 from snowflake.cli._plugins.dcm.reporters import (
     AnalyzeReporter,
     PlanReporter,
@@ -336,24 +337,26 @@ def deploy(
     context = _resolve_context_with_required_manifest(from_location, identifier, target)
     project_id = context.project_identifier
 
-    manager = DCMProjectManager()
-    effective_stage = manager.sync_local_files(
-        project_identifier=project_id,
-        source_directory=str(from_location.path),
-    )
+    if skip_plan:
+        cli_console.warning("Skipping planning step")
 
-    with cli_console.spinner() as spinner:
-        spinner.add_task(description=f"Deploying dcm project {project_id}", total=None)
-        if skip_plan:
-            cli_console.warning("Skipping planning step")
-        result = manager.deploy(
+    manager = DCMProjectManager()
+    tracker = DeployProgressTracker(conn=manager.connection)
+    with tracker.session():
+        effective_stage = manager.sync_local_files(
+            project_identifier=project_id,
+            source_directory=str(from_location.path),
+            progress=tracker,
+        )
+        sfqid = manager.deploy_async(
             project_identifier=project_id,
             configuration=context.configuration,
             from_stage=effective_stage,
             variables=variables,
             alias=alias,
             skip_plan=skip_plan,
         )
+        result = tracker.run_deploy_poll(sfqid)
 
     reporter = PlanReporter(save_output=save_output, command_name="deploy")
     return reporter.process(result)
@@ -422,15 +425,18 @@ def purge(
     if not force:
         _confirm_purge(project_id)
 
-    with cli_console.spinner() as spinner:
-        spinner.add_task(description=f"Purging dcm project {project_id}", total=None)
-        if skip_plan:
-            cli_console.warning("Skipping planning step")
-        result = DCMProjectManager().purge(
+    if skip_plan:
+        cli_console.warning("Skipping planning step")
+
+    manager = DCMProjectManager()
+    tracker = DeployProgressTracker(conn=manager.connection, operation="purge")
+    with tracker.session():
+        sfqid = manager.purge_async(
             project_identifier=project_id,
             alias=alias,
             skip_plan=skip_plan,
         )
+        result = tracker.run_deploy_poll(sfqid)
 
     reporter = PlanReporter(save_output=save_output, command_name="purge")
     return reporter.process(result)
@@ -461,20 +467,24 @@ def plan(
     project_id = context.project_identifier
 
     manager = DCMProjectManager()
-    effective_stage = manager.sync_local_files(
-        project_identifier=project_id,
-        source_directory=str(from_location.path),
-    )
-
-    with cli_console.spinner() as spinner:
-        spinner.add_task(description=f"Planning dcm project {project_id}", total=None)
-        result = manager.plan(
+    tracker = DeployProgressTracker(conn=manager.connection, operation="plan")
+    with tracker.session():
+        effective_stage = manager.sync_local_files(
             project_identifier=project_id,
-            configuration=context.configuration,
-            from_stage=effective_stage,
-            variables=variables,
-            save_output=save_output,
-            delta=delta,
+            source_directory=str(from_location.path),
+            progress=tracker,
+        )
+        result = tracker.run_loader_phase(
+            lambda: manager.plan(
+                project_identifier=project_id,
+                configuration=context.configuration,
+                from_stage=effective_stage,
+                variables=variables,
+                save_output=save_output,
+                delta=delta,
+            ),
+            phase_name="PLAN",
+            simulated_phases=("RENDER", "COMPILE"),
         )
 
     reporter = PlanReporter(save_output=save_output, command_name="plan")

src/snowflake/cli/_plugins/dcm/manager.py

@@ -59,6 +59,24 @@ def connection(self):
         """Exposes the underlying Snowflake connection."""
         return self._conn
 
+    def _build_deploy_query(
+        self,
+        project_identifier: FQN,
+        from_stage: str,
+        configuration: str | None,
+        variables: List[str] | None,
+        alias: str | None,
+        skip_plan: bool,
+    ) -> str:
+        query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} DEPLOY"
+        if alias:
+            query += f' AS "{alias}"'
+        query += self._get_configuration_and_variables_query(configuration, variables)
+        query += self._get_from_stage_query(from_stage)
+        if skip_plan:
+            query += " SKIP PLAN"
+        return query
+
     def deploy(
         self,
         project_identifier: FQN,
@@ -75,15 +93,48 @@ def deploy(
             len(variables or []),
             skip_plan,
         )
-        query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} DEPLOY"
-        if alias:
-            query += f' AS "{alias}"'
-        query += self._get_configuration_and_variables_query(configuration, variables)
-        query += self._get_from_stage_query(from_stage)
-        if skip_plan:
-            query += f" SKIP PLAN"
+        query = self._build_deploy_query(
+            project_identifier, from_stage, configuration, variables, alias, skip_plan
+        )
         return self.execute_query(query=query)
 
+    def deploy_async(
+        self,
+        project_identifier: FQN,
+        from_stage: str,
+        configuration: str | None = None,
+        variables: List[str] | None = None,
+        alias: str | None = None,
+        skip_plan: bool = False,
+    ) -> str:
+        """
+        Submits a deploy query asynchronously and returns the Snowflake query ID (sfqid).
+        Use with :class:`~snowflake.cli._plugins.dcm.progress.DeployProgressTracker`
+        to poll progress and obtain the final result cursor.
+        """
+        log.info(
+            "Submitting DCM deploy async (project_identifier=%s, has_configuration=%s, variables_count=%d, skip_plan=%s).",
+            project_identifier,
+            bool(configuration),
+            len(variables or []),
+            skip_plan,
+        )
+        query = self._build_deploy_query(
+            project_identifier, from_stage, configuration, variables, alias, skip_plan
+        )
+        # Closing the cursor does not cancel the async query; it keeps running
+        # server-side and its results are fetched later via the sfqid (see
+        # DeployProgressTracker.run_deploy_poll -> get_results_from_sfqid).
+        with self._conn.cursor() as cursor:
+            cursor.execute_async(query)
+            sfqid = cursor.sfqid
+        log.info(
+            "DCM deploy async submitted (project_identifier=%s, sfqid=%s).",
+            project_identifier,
+            sfqid,
+        )
+        return sfqid
+
     def raw_analyze(
         self,
         project_identifier: FQN,
@@ -223,12 +274,50 @@ def purge(
             project_identifier,
             skip_plan,
         )
+        query = self._build_purge_query(project_identifier, alias, skip_plan)
+        return self.execute_query(query=query)
+
+    def purge_async(
+        self,
+        project_identifier: FQN,
+        alias: str | None = None,
+        skip_plan: bool = False,
+    ) -> str:
+        """
+        Submits a purge query asynchronously and returns the Snowflake query ID (sfqid).
+        Use with :class:`~snowflake.cli._plugins.dcm.progress.DeployProgressTracker`
+        to poll progress and obtain the final result cursor.
+        """
+        log.info(
+            "Submitting DCM purge async (project_identifier=%s, skip_plan=%s).",
+            project_identifier,
+            skip_plan,
+        )
+        query = self._build_purge_query(project_identifier, alias, skip_plan)
+        # Closing the cursor does not cancel the async query; results are
+        # fetched later via the sfqid (see run_deploy_poll).
+        with self._conn.cursor() as cursor:
+            cursor.execute_async(query)
+            sfqid = cursor.sfqid
+        log.info(
+            "DCM purge async submitted (project_identifier=%s, sfqid=%s).",
+            project_identifier,
+            sfqid,
+        )
+        return sfqid
+
+    @staticmethod
+    def _build_purge_query(
+        project_identifier: FQN,
+        alias: str | None,
+        skip_plan: bool,
+    ) -> str:
         query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} PURGE"
         if alias:
             query += f' AS "{alias}"'
         if skip_plan:
             query += " SKIP PLAN"
-        return self.execute_query(query=query)
+        return query
 
     def test(self, project_identifier: FQN) -> SnowflakeCursor:
         log.info(

src/snowflake/cli/_plugins/dcm/progress.py

@@ -457,14 +457,14 @@ def _append_progress_bar(self, out: Text, progress: int) -> None:
         in_progress = 0 < filled < _BAR_WIDTH
 
         if in_progress:
-            out.append(_BAR_CELL * filled + _BAR_LEADING_EDGE, style="blue")
+            out.append(_BAR_CELL * filled + _BAR_LEADING_EDGE, style=styles.BLUE)
             out.append(_BAR_CELL * (_BAR_WIDTH - filled - 1), style="dim")
         elif filled == 0:
             out.append(_BAR_CELL * _BAR_WIDTH, style="dim")
         else:
-            out.append(_BAR_CELL * _BAR_WIDTH, style="blue")
+            out.append(_BAR_CELL * _BAR_WIDTH, style=styles.BLUE)
 
-        out.append(f" {progress:>3}%", style="blue")
+        out.append(f" {progress:>3}%", style=styles.BLUE)
 
     def _append_upload_details(self, out: Text) -> None:
         """Render the stage-creation message and folder counters indented
@@ -509,7 +509,7 @@ def _render_phase_line(self, out: Text, phase: _Phase) -> None:
                 # Blue matches the active-indicator color used by the
                 # pip-style progress bar so all "this phase is in flight"
                 # signals share the same hue.
-                out.append(_spinner_glyph(), style="blue")
+                out.append(_spinner_glyph(), style=styles.BLUE)
             out.append(duration_str + "\n", style="dim")
         else:  # PENDING
             out.append(name_col + "·\n", style="dim")

src/snowflake/cli/_plugins/dcm/styles.py

@@ -13,13 +13,17 @@
 # limitations under the License.
 from rich.style import Style
 
-_COLOR_BLUE = "#a0a8fe"
+# Single source of truth for the DCM "blue" hue. Use the terminal-default
+# named color (not a fixed hex) so it respects the user's theme and stays
+# identical everywhere it's used: progress spinner/bar, running phase,
+# refresh status, and unknown rows.
+BLUE = "blue"
 
 DOMAIN_STYLE = Style(color="cyan")
 BOLD_STYLE = Style(bold=True)
 
 # Refresh
-STATUS_STYLE = Style(color=_COLOR_BLUE)
+STATUS_STYLE = Style(color=BLUE)
 REMOVED_STYLE = Style(color="red", italic=True)
 INSERTED_STYLE = Style(color="green", italic=True)
 
@@ -31,9 +35,9 @@
 CREATE_STYLE = Style(color="green")
 ALTER_STYLE = Style(color="yellow")
 DROP_STYLE = Style(color="red")
-UNKNOWN_STYLE = Style(color=_COLOR_BLUE)
+UNKNOWN_STYLE = Style(color=BLUE)
 
 # Deploy progress phases
 PHASE_DONE_STYLE = Style(color="green", bold=True)
-PHASE_RUNNING_STYLE = Style(color="blue", bold=True)
+PHASE_RUNNING_STYLE = Style(color=BLUE, bold=True)
 PHASE_FAILED_STYLE = Style(color="red", bold=True)