feat: [SNOW-2326971] implement dcm preview command

Author: sfc-gh-jwilkowski

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

@@ -24,6 +24,7 @@
 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,
@@ -127,13 +128,12 @@ def deploy(
 
     with cli_console.spinner() as spinner:
         spinner.add_task(description=f"Deploying dcm project {identifier}", total=None)
-        result = manager.execute(
+        result = manager.deploy(
             project_identifier=identifier,
             configuration=configuration,
             from_stage=effective_stage,
             variables=variables,
             alias=alias,
-            output_path=None,
         )
     return QueryJsonValueResult(result)
 
@@ -157,11 +157,10 @@ def plan(
 
     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,
         )
@@ -308,6 +307,53 @@ def refresh(
     return MessageResult(message)
 
 
+@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)
+
+
 def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
     manager = DCMProjectManager()
     if not from_location:

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

@@ -77,36 +77,37 @@ def _collect_output(
         else:
             cli_console.step(f"Plan output saved to: {output_path}")
 
-    def execute(
+    def deploy(
         self,
         project_identifier: FQN,
         from_stage: str,
         configuration: str | None = None,
         variables: List[str] | None = None,
-        dry_run: bool = False,
         alias: str | None = None,
+    ):
+        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)
+        return self.execute_query(query=query)
+
+    def plan(
+        self,
+        project_identifier: FQN,
+        from_stage: str,
+        configuration: str | None = None,
+        variables: List[str] | None = None,
         output_path: str | None = None,
     ):
-        with self._collect_output(project_identifier, output_path) if (
-            output_path and dry_run
-        ) else nullcontext() as output_stage:
-            query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier}"
-            if dry_run:
-                query += " PLAN"
-            else:
-                query += " DEPLOY"
-                if alias:
-                    query += f' AS "{alias}"'
-            if configuration or variables:
-                query += f" USING"
-            if configuration:
-                query += f" CONFIGURATION {configuration}"
-            if variables:
-                query += StageManager.parse_execute_variables(
-                    parse_key_value_variables(variables)
-                ).removeprefix(" using")
-            stage_path = StagePath.from_stage_str(from_stage)
-            query += f" FROM {stage_path.absolute_path()}"
+        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} PLAN"
+            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)
@@ -144,6 +145,42 @@ def refresh(self, project_identifier: FQN):
         query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} REFRESH ALL"
         return self.execute_query(query=query)
 
+    def preview(
+        self,
+        project_identifier: FQN,
+        object_identifier: FQN,
+        from_stage: str,
+        configuration: str | None = None,
+        variables: List[str] | None = None,
+        limit: int | None = None,
+    ):
+        query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} PREVIEW {object_identifier.sql_identifier}"
+        query += self._get_configuration_and_variables_query(configuration, variables)
+        query += self._get_from_stage_query(from_stage)
+        if limit is not None:
+            query += f" LIMIT {limit}"
+        return self.execute_query(query=query)
+
+    @staticmethod
+    def _get_from_stage_query(from_stage: str) -> str:
+        stage_path = StagePath.from_stage_str(from_stage)
+        return f" FROM {stage_path.absolute_path()}"
+
+    @staticmethod
+    def _get_configuration_and_variables_query(
+        configuration: str | None, variables: List[str] | None
+    ) -> str:
+        query = ""
+        if configuration or variables:
+            query += f" USING"
+        if configuration:
+            query += f" CONFIGURATION {configuration}"
+        if variables:
+            query += StageManager.parse_execute_variables(
+                parse_key_value_variables(variables)
+            ).removeprefix(" using")
+        return query
+
     @staticmethod
     def sync_local_files(
         project_identifier: FQN, source_directory: str | None = None