Add snow dcm compile and dependencies commands

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

@@ -27,11 +27,15 @@
 from snowflake.cli._plugins.dcm.reporters import (
     AnalyzeErrorsReporter,
     AnalyzeReporter,
+    DependenciesReporter,
     PlanReporter,
     RefreshReporter,
     TestReporter,
 )
 from snowflake.cli._plugins.dcm.utils import (
+    RENDERED_DEFINITIONS_FOLDER,
+    announce_compile_separator,
+    announce_rendered_definitions,
     clear_command_artifacts,
     mock_dcm_response,
 )
@@ -528,10 +532,11 @@ def raw_analyze(
 
 
 @app.command(
+    name="compile",
     requires_connection=True,
     hidden=not FeatureFlag.ENABLE_DCM_EARLY_ACCESS.is_enabled(),
 )
-def analyze_errors(
+def compile_project(
     identifier: Optional[FQN] = optional_dcm_identifier,
     from_location: SecurePath = from_option,
     variables: Optional[List[str]] = variables_flag,
@@ -540,9 +545,9 @@ def analyze_errors(
     **options,
 ):
     """
-    Analyzes a DCM Project and prints a formatted list of errors found.
+    Compiles a DCM Project and prints a formatted list of errors found.
     """
-    clear_command_artifacts("analyze-errors")
+    clear_command_artifacts("compile", folder_name=RENDERED_DEFINITIONS_FOLDER)
 
     context = _resolve_context_with_required_manifest(from_location, identifier, target)
     project_id = context.project_identifier
@@ -562,14 +567,75 @@ def analyze_errors(
                 from_stage=effective_stage,
                 variables=variables,
                 save_output=save_output,
-                command_name="analyze-errors",
+                command_name="compile",
+                output_folder_name=RENDERED_DEFINITIONS_FOLDER,
             ),
             phase_name="COMPILE",
             simulated_phases=["RENDER"],
         )
 
     reporter = AnalyzeErrorsReporter(save_output=save_output)
-    return reporter.process(result)
+    if save_output:
+        announce_rendered_definitions()
+    try:
+        return reporter.process(result)
+    finally:
+        announce_compile_separator()
+
+
+@app.command(
+    requires_connection=True,
+    hidden=not FeatureFlag.ENABLE_DCM_EARLY_ACCESS.is_enabled(),
+)
+def dependencies(
+    identifier: Optional[FQN] = optional_dcm_identifier,
+    from_location: SecurePath = from_option,
+    variables: Optional[List[str]] = variables_flag,
+    target: Optional[str] = target_option,
+    save_output: bool = save_output_option,
+    **options,
+):
+    """
+    Analyzes a DCM Project and generates a dependency diagram.
+
+    The diagram is written as a Mermaid flowchart in a Markdown file that can
+    be opened in your IDE's Markdown preview to explore object dependencies.
+    """
+    clear_command_artifacts("dependencies", folder_name=RENDERED_DEFINITIONS_FOLDER)
+
+    context = _resolve_context_with_required_manifest(from_location, identifier, target)
+    project_id = context.project_identifier
+
+    manager = DCMProjectManager()
+    tracker = DeployProgressTracker(conn=manager.connection, operation="compile")
+    with tracker.session():
+        effective_stage = manager.sync_local_files(
+            project_identifier=project_id,
+            source_directory=str(from_location.path),
+            progress=tracker,
+        )
+        result = tracker.run_loader_phase(
+            lambda: manager.raw_analyze(
+                project_identifier=project_id,
+                configuration=context.configuration,
+                from_stage=effective_stage,
+                variables=variables,
+                save_output=save_output,
+                command_name="dependencies",
+                output_folder_name=RENDERED_DEFINITIONS_FOLDER,
+            ),
+            phase_name="COMPILE",
+            simulated_phases=["RENDER"],
+        )
+
+    reporter = DependenciesReporter(
+        project_identifier=project_id, save_output=save_output
+    )
+    try:
+        return reporter.process(result)
+    finally:
+        if save_output:
+            announce_rendered_definitions()
 
 
 @app.command(requires_connection=True)

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

@@ -143,6 +143,7 @@ def raw_analyze(
         variables: List[str] | None = None,
         save_output: bool = False,
         command_name: str = "raw-analyze",
+        output_folder_name: str | None = None,
     ):
         log.info(
             "Running DCM analyze manager operation (command_name=%s, project_identifier=%s, has_configuration=%s, variables_count=%d, save_output=%s).",
@@ -158,7 +159,9 @@ def raw_analyze(
 
         if save_output:
             with collect_output(
-                project_identifier, command_name=command_name
+                project_identifier,
+                command_name=command_name,
+                folder_name=output_folder_name,
             ) as output_stage:
                 query += f" OUTPUT_PATH {output_stage}"
                 result = self.execute_query(query=query)

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/reporters/__init__.py

@@ -16,13 +16,15 @@
     AnalyzeReporter,
 )
 from snowflake.cli._plugins.dcm.reporters.base import Reporter
+from snowflake.cli._plugins.dcm.reporters.dependencies import DependenciesReporter
 from snowflake.cli._plugins.dcm.reporters.plan import PlanReporter
 from snowflake.cli._plugins.dcm.reporters.refresh import RefreshReporter
 from snowflake.cli._plugins.dcm.reporters.test import TestReporter
 
 __all__ = [
     "AnalyzeErrorsReporter",
     "AnalyzeReporter",
+    "DependenciesReporter",
     "Reporter",
     "PlanReporter",
     "RefreshReporter",

src/snowflake/cli/_plugins/dcm/reporters/analyze.py

@@ -322,7 +322,10 @@ class AnalyzeErrorsReporter(Reporter[_FileFindings]):
 
     def __init__(self, save_output: bool = False):
         super().__init__(save_output=save_output)
-        self.command_name = "analyze-errors"
+        self.command_name = "compile"
+        # ``compile`` prints its own "Rendered definitions saved to:" line, so
+        # suppress the generic "Artifacts saved to" step.
+        self.announce_save = False
         self._error_count = 0
         self._warning_count = 0
         self._info_count = 0

src/snowflake/cli/_plugins/dcm/reporters/base.py

@@ -40,6 +40,10 @@ def __init__(self, save_output: bool = False) -> None:
         self.result_raw_data = None
         self.command_name = ""
         self.save_output = save_output
+        # When False, saving the raw response won't print the "Artifacts saved
+        # to" step (the file is still written). Commands that render their own
+        # output-location line (e.g. ``compile``) opt out of the default step.
+        self.announce_save = True
 
     @abstractmethod
     def extract_data(self, result_json: Dict[str, Any]) -> List[Dict[str, Any]]:
@@ -77,7 +81,11 @@ def print_summary(self) -> None:
     def _try_save_response(self, result_json: Dict[str, Any]) -> None:
         """Save raw JSON response if save_output is enabled and raw data is available."""
         if self.save_output:
-            save_command_response(self.command_name, result_json)
+            save_command_response(
+                self.command_name,
+                result_json,
+                announce=self.announce_save,
+            )
 
     def process_payload(self, result_json: Dict[str, Any]) -> None:
         """Process already decoded response payload and print results."""