Jw/rich dcm reporting

.pre-commit-config.yaml

@@ -36,7 +36,7 @@ repos:
       - id: check-print-in-code
         language: pygrep
         name: "Check for print statements"
-        entry: "print\\(|echo\\("
+        entry: "\\bprint\\(|echo\\("
         pass_filenames: true
         files: ^src/snowflake/.*\.py$
         exclude: >

src/snowflake/cli/_app/printing.py

@@ -28,6 +28,7 @@
 from rich import print as rich_print
 from rich.live import Live
 from rich.table import Table
+from snowflake.cli._plugins.dcm.reporting import DCMMessageResult
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.output.formats import OutputFormat
 from snowflake.cli.api.output.types import (
@@ -317,6 +318,8 @@ def print_unstructured(obj: CommandResult | None):
         rich_print("Done", flush=True)
     elif not obj.result:
         rich_print("No data", flush=True)
+    elif isinstance(obj, DCMMessageResult):
+        rich_print(obj.message)
     elif isinstance(obj, MessageResult):
         rich_print(sanitize_for_terminal(obj.message), flush=True)
     else:

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

@@ -11,14 +11,27 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from pathlib import Path
 from typing import List, Optional
 
 import typer
-from snowflake.cli._plugins.dcm.manager import DCMProjectManager
+from snowflake.cli._plugins.dcm.debug import get_debug_cursor
+from snowflake.cli._plugins.dcm.manager import AnalysisType, DCMProjectManager
+from snowflake.cli._plugins.dcm.reporting import (
+    AnalyzeReporter,
+    DCMCommandResult,
+    PlanReporter,
+    RefreshReporter,
+    TestReporter,
+)
+from snowflake.cli._plugins.dcm.utils import (
+    TestResultFormat,
+)
 from snowflake.cli._plugins.object.command_aliases import add_object_command_aliases
 from snowflake.cli._plugins.object.commands import scope_option
 from snowflake.cli._plugins.object.manager import ObjectManager
 from snowflake.cli.api.commands.flags import (
+    IdentifierType,
     IfExistsOption,
     IfNotExistsOption,
     OverrideableOption,
@@ -36,7 +49,6 @@
 from snowflake.cli.api.identifiers import FQN
 from snowflake.cli.api.output.types import (
     MessageResult,
-    QueryJsonValueResult,
     QueryResult,
 )
 from snowflake.cli.api.utils.path_utils import is_stage_path
@@ -47,6 +59,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>"`.'
@@ -112,25 +125,38 @@ def deploy(
     variables: Optional[List[str]] = variables_flag,
     configuration: Optional[str] = configuration_flag,
     alias: Optional[str] = alias_option,
+    skip_plan: bool = typer.Option(
+        False,
+        "--skip-plan",
+        help="Skips planning step",
+    ),
     **options,
 ):
     """
     Applies changes defined in DCM Project to Snowflake.
     """
+    # Check for debug mode
+    debug_cursor = get_debug_cursor("deploy")
+    if debug_cursor:
+        return DCMCommandResult(debug_cursor, PlanReporter("deploy")).process()
+
     manager = DCMProjectManager()
     effective_stage = _get_effective_stage(identifier, from_location)
 
     with cli_console.spinner() as spinner:
         spinner.add_task(description=f"Deploying dcm project {identifier}", total=None)
-        result = manager.execute(
+        if skip_plan:
+            cli_console.warning("Skipping planning step")
+        result = manager.deploy(
             project_identifier=identifier,
             configuration=configuration,
             from_stage=effective_stage,
             variables=variables,
             alias=alias,
-            output_path=None,
+            skip_plan=skip_plan,
         )
-    return QueryJsonValueResult(result)
+
+    return DCMCommandResult(result, PlanReporter("deploy")).process()
 
 
 @app.command(requires_connection=True)
@@ -147,21 +173,25 @@ def plan(
     """
     Plans a DCM Project deployment (validates without executing).
     """
+    # Check for debug mode
+    debug_cursor = get_debug_cursor("plan")
+    if debug_cursor:
+        return DCMCommandResult(debug_cursor, PlanReporter("plan")).process()
+
     manager = DCMProjectManager()
     effective_stage = _get_effective_stage(identifier, from_location)
 
     with cli_console.spinner() as spinner:
         spinner.add_task(description=f"Planning dcm project {identifier}", total=None)
-        result = manager.execute(
+        result = manager.plan(
             project_identifier=identifier,
             configuration=configuration,
             from_stage=effective_stage,
-            dry_run=True,
             variables=variables,
             output_path=output_path,
         )
 
-    return QueryJsonValueResult(result)
+    return DCMCommandResult(result, PlanReporter("plan")).process()
 
 
 @app.command(requires_connection=True)
@@ -235,14 +265,174 @@ def drop_deployment(
     )
 
 
-def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
+@app.command(requires_connection=True)
+def test(
+    identifier: FQN = dcm_identifier,
+    export_format: Optional[List[TestResultFormat]] = typer.Option(
+        None,
+        "--result-format",
+        help="Export test results in specified format(s) into directory set with `--output-path`. Can be specified multiple times for multiple formats.",
+        show_default=False,
+    ),
+    output_path: Optional[Path] = typer.Option(
+        None,
+        "--output-path",
+        help="Directory where test result files will be saved. Defaults to current directory.",
+        show_default=False,
+    ),
+    **options,
+):
+    """
+    Test all expectations set for tables, views and dynamic tables defined
+    in DCM project.
+    """
+    # Check for debug mode
+    debug_cursor = get_debug_cursor("test")
+    if debug_cursor:
+        return DCMCommandResult(debug_cursor, TestReporter()).process()
+
+    with cli_console.spinner() as spinner:
+        spinner.add_task(description=f"Testing dcm project {identifier}", total=None)
+        result = DCMProjectManager().test(project_identifier=identifier)
+
+    # TODO: Integrate export_format and output_path into TestReporter
+    # The export_test_results function (JUnit, TAP, JSON export) should be
+    # integrated into the reporter's process() method or added as a post-process hook.
+    # This would allow the reporter to handle all output formats consistently.
+    # For now, export_format functionality is disabled during refactoring.
+
+    return DCMCommandResult(result, TestReporter()).process()
+
+
+@app.command(requires_connection=True)
+def refresh(
+    identifier: FQN = dcm_identifier,
+    **options,
+):
+    """
+    Refreshes dynamic tables defined in DCM project.
+    """
+    # Check for debug mode
+    debug_cursor = get_debug_cursor("refresh")
+    if debug_cursor:
+        return DCMCommandResult(debug_cursor, RefreshReporter()).process()
+
+    with cli_console.spinner() as spinner:
+        spinner.add_task(description=f"Refreshing dcm project {identifier}", total=None)
+        result = DCMProjectManager().refresh(project_identifier=identifier)
+
+    return DCMCommandResult(result, RefreshReporter()).process()
+
+
+@app.command(requires_connection=True)
+def preview(
+    identifier: FQN = dcm_identifier,
+    object_identifier: FQN = typer.Option(
+        ...,
+        "--object",
+        help="FQN of table/view/etc to be previewed.",
+        show_default=False,
+        click_type=IdentifierType(),
+    ),
+    from_location: Optional[str] = from_option,
+    variables: Optional[List[str]] = variables_flag,
+    configuration: Optional[str] = configuration_flag,
+    limit: Optional[int] = typer.Option(
+        None,
+        "--limit",
+        help="The maximum number of rows to be returned.",
+        show_default=False,
+    ),
+    **options,
+):
+    """
+    Returns rows from any table, view, dynamic table.
+
+    Examples:
+    \nsnow dcm preview MY_PROJECT --configuration DEV --object MY_DB.PUBLIC.MY_VIEW --limit 2
+    """
+    manager = DCMProjectManager()
+    effective_stage = _get_effective_stage(identifier, from_location)
+
+    with cli_console.spinner() as spinner:
+        spinner.add_task(
+            description=f"Previewing {object_identifier}.",
+            total=None,
+        )
+        result = manager.preview(
+            project_identifier=identifier,
+            object_identifier=object_identifier,
+            configuration=configuration,
+            from_stage=effective_stage,
+            variables=variables,
+            limit=limit,
+        )
+
+    return QueryResult(result)
+
+
+@app.command(requires_connection=True)
+def analyze(
+    identifier: FQN = dcm_identifier,
+    from_location: Optional[str] = from_option,
+    files: Optional[List[Path]] = typer.Option(
+        None,
+        "--files",
+        help="Files to analyze. Can be specified multiple times for multiple files.",
+        show_default=False,
+    ),
+    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.
+    """
+    # Check for debug mode
+    debug_cursor = get_debug_cursor("analyze")
+    if debug_cursor:
+        return DCMCommandResult(debug_cursor, AnalyzeReporter()).process()
+
+    manager = DCMProjectManager()
+    effective_stage = _get_effective_stage(identifier, from_location, files)
+
+    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,
+            files=files,
+        )
+
+    return DCMCommandResult(result, AnalyzeReporter()).process()
+
+
+def _get_effective_stage(
+    identifier: FQN, from_location: Optional[str], files: Optional[List[Path]] = None
+):
     manager = DCMProjectManager()
     if not from_location:
-        from_stage = manager.sync_local_files(project_identifier=identifier)
+        from_stage = manager.sync_local_files(
+            project_identifier=identifier, files=files
+        )
     elif is_stage_path(from_location):
         from_stage = from_location
     else:
         from_stage = manager.sync_local_files(
-            project_identifier=identifier, source_directory=from_location
+            project_identifier=identifier, source_directory=from_location, files=files
         )
     return from_stage