feat: [SNOW-2412813] wip: analyze

Author: sfc-gh-jwilkowski

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

@@ -12,9 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
-from typing import List, Optional
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
 
 import typer
+from snowflake.cli._plugins.dcm.manager import AnalysisType
 from snowflake.cli._plugins.dcm.manager import DCMProjectManager
 from snowflake.cli._plugins.dcm.utils import (
     format_refresh_results,
@@ -53,6 +55,7 @@
     is_hidden=FeatureFlag.ENABLE_SNOWFLAKE_PROJECTS.is_disabled,
 )
 
+
 dcm_identifier = identifier_argument(sf_object="DCM Project", example="MY_PROJECT")
 variables_flag = variables_option(
     'Variables for the execution context; for example: `-D "<key>=<value>"`.'
@@ -354,6 +357,124 @@ def preview(
     return QueryResult(result)
 
 
+@app.command(requires_connection=True)
+def analyze(
+    identifier: FQN = dcm_identifier,
+    from_location: Optional[str] = from_option,
+    variables: Optional[List[str]] = variables_flag,
+    configuration: Optional[str] = configuration_flag,
+    analysis_type: Optional[AnalysisType] = typer.Option(
+        None,
+        "--type",
+        help="Type of analysis to perform.",
+        show_default=False,
+        case_sensitive=False,
+    ),
+    output_path: Optional[str] = output_path_option(
+        help="Path where the analysis result will be stored. Can be a stage path (starting with '@') or a local directory path."
+    ),
+    **options,
+):
+    """
+    Analyzes a DCM Project.
+    """
+    manager = DCMProjectManager()
+    effective_stage = _get_effective_stage(identifier, from_location)
+
+    with cli_console.spinner() as spinner:
+        spinner.add_task(description=f"Analyzing dcm project {identifier}", total=None)
+        result = manager.analyze(
+            project_identifier=identifier,
+            configuration=configuration,
+            from_stage=effective_stage,
+            variables=variables,
+            analysis_type=analysis_type,
+            output_path=output_path,
+        )
+
+    row = result.fetchone()
+    if not row:
+        return MessageResult("No data.")
+
+    result_data = row[0]
+    result_json = (
+        json.loads(result_data) if isinstance(result_data, str) else result_data
+    )
+
+    summary = _analyze_result_summary(result_json)
+
+    if summary.has_errors:
+        error_message = _format_error_message(summary)
+        raise CliError(error_message)
+
+    return MessageResult(
+        f"✓ Analysis complete: {summary.total_files} file(s) analyzed, "
+        f"{summary.total_definitions} definition(s) found. No errors detected."
+    )
+
+
+@dataclass
+class AnalysisSummary:
+    total_files: int = 0
+    total_definitions: int = 0
+    files_with_errors: int = 0
+    total_errors: int = 0
+    errors_by_file: Dict[str, List[str]] = field(default_factory=dict)
+    has_errors: bool = False
+
+
+def _analyze_result_summary(result_json) -> AnalysisSummary:
+    summary = AnalysisSummary()
+
+    if not isinstance(result_json, dict):
+        return summary
+
+    files = result_json.get("files", [])
+    summary.total_files = len(files)
+
+    for file_info in files:
+        source_path = file_info.get("sourcePath", "unknown")
+        file_errors = []
+
+        # Check file-level errors
+        for error in file_info.get("errors", []):
+            error_msg = error.get("message", "Unknown error")
+            file_errors.append(error_msg)
+            summary.total_errors += 1
+
+        # Check definition-level errors
+        definitions = file_info.get("definitions", [])
+        summary.total_definitions += len(definitions)
+
+        for definition in definitions:
+            for error in definition.get("errors", []):
+                error_msg = error.get("message", "Unknown error")
+                file_errors.append(error_msg)
+                summary.total_errors += 1
+
+        if file_errors:
+            summary.errors_by_file[source_path] = file_errors
+            summary.files_with_errors += 1
+            summary.has_errors = True
+
+    return summary
+
+
+def _format_error_message(summary: AnalysisSummary) -> str:
+    lines = [
+        f"Analysis found {summary.total_errors} error(s) in {summary.files_with_errors} file(s):",
+        "",
+    ]
+
+    for file_path, errors in summary.errors_by_file.items():
+        lines.append(f"  {file_path}:")
+        for error in errors:
+            lines.append(f"    • {error}")
+        lines.append("")
+
+    return "\n".join(lines).rstrip()
+
+
 def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
     manager = DCMProjectManager()
     if not from_location:

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

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from contextlib import contextmanager, nullcontext
+from enum import Enum
 from pathlib import Path
 from typing import Generator, List
 
@@ -38,6 +39,10 @@
 DCM_PROJECT_TYPE = "dcm_project"
 
 
+class AnalysisType(str, Enum):
+    DEPENDENCIES = "dependencies"
+
+
 class DCMProjectManager(SqlExecutionMixin):
     @contextmanager
     def _collect_output(
@@ -161,6 +166,31 @@ def preview(
             query += f" LIMIT {limit}"
         return self.execute_query(query=query)
 
+    def analyze(
+        self,
+        project_identifier: FQN,
+        from_stage: str,
+        configuration: str | None = None,
+        variables: List[str] | None = None,
+        analysis_type: AnalysisType | None = None,
+        output_path: str | None = None,
+    ):
+        with self._collect_output(
+            project_identifier, output_path
+        ) if output_path else nullcontext() as output_stage:
+            query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} ANALYZE"
+            if analysis_type:
+                query += f" {analysis_type.value.upper()}"
+            query += self._get_configuration_and_variables_query(
+                configuration, variables
+            )
+            query += self._get_from_stage_query(from_stage)
+            if output_stage is not None:
+                query += f" OUTPUT_PATH {output_stage}"
+            result = self.execute_query(query=query)
+
+        return result
+
     @staticmethod
     def _get_from_stage_query(from_stage: str) -> str:
         stage_path = StagePath.from_stage_str(from_stage)