Workspaces application package - drop action and command (#1517)

* app package drop

* integration tests

src/snowflake/cli/_plugins/nativeapp/teardown_processor.py

@@ -22,13 +22,8 @@
 from snowflake.cli._plugins.nativeapp.constants import (
     ALLOWED_SPECIAL_COMMENTS,
     COMMENT_COL,
-    EXTERNAL_DISTRIBUTION,
-    INTERNAL_DISTRIBUTION,
     OWNER_COL,
 )
-from snowflake.cli._plugins.nativeapp.exceptions import (
-    CouldNotDropApplicationPackageWithVersions,
-)
 from snowflake.cli._plugins.nativeapp.manager import (
     NativeAppCommandProcessor,
     NativeAppManager,
@@ -37,11 +32,15 @@
     needs_confirmation,
 )
 from snowflake.cli.api.console import cli_console as cc
-from snowflake.cli.api.entities.utils import ensure_correct_owner
+from snowflake.cli.api.entities.application_package_entity import (
+    ApplicationPackageEntity,
+)
+from snowflake.cli.api.entities.utils import (
+    drop_generic_object,
+    ensure_correct_owner,
+)
 from snowflake.cli.api.errno import APPLICATION_NO_LONGER_AVAILABLE
-from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
 from snowflake.connector import ProgrammingError
-from snowflake.connector.cursor import DictCursor
 
 
 class NativeAppTeardownProcessor(NativeAppManager, NativeAppCommandProcessor):
@@ -51,20 +50,13 @@ def __init__(self, project_definition: Dict, project_root: Path):
     def drop_generic_object(
         self, object_type: str, object_name: str, role: str, cascade: bool = False
     ):
-        """
-        Drop object using the given role.
-        """
-        with self.use_role(role):
-            cc.step(f"Dropping {object_type} {object_name} now.")
-            drop_query = f"drop {object_type} {object_name}"
-            if cascade:
-                drop_query += " cascade"
-            try:
-                self._execute_query(drop_query)
-            except:
-                raise SnowflakeSQLExecutionError(drop_query)
-
-            cc.message(f"Dropped {object_type} {object_name} successfully.")
+        return drop_generic_object(
+            console=cc,
+            object_type=object_type,
+            object_name=object_name,
+            role=role,
+            cascade=cascade,
+        )
 
     def drop_application(
         self, auto_yes: bool, interactive: bool = False, cascade: Optional[bool] = None
@@ -198,90 +190,12 @@ def drop_application(
         return  # The application object was successfully dropped, therefore exit gracefully
 
     def drop_package(self, auto_yes: bool):
-        """
-        Attempts to drop application package unless user specifies otherwise.
-        """
-        needs_confirm = True
-
-        # 1. If existing application package is not found, exit gracefully
-        show_obj_row = self.get_existing_app_pkg_info()
-        if show_obj_row is None:
-            cc.warning(
-                f"Role {self.package_role} does not own any application package with the name {self.package_name}, or the application package does not exist."
-            )
-            return
-
-        # 2. Check for the right owner
-        ensure_correct_owner(
-            row=show_obj_row, role=self.package_role, obj_name=self.package_name
-        )
-
-        with self.use_role(self.package_role):
-            # 3. Check for versions in the application package
-            show_versions_query = (
-                f"show versions in application package {self.package_name}"
-            )
-            show_versions_cursor = self._execute_query(
-                show_versions_query, cursor_class=DictCursor
-            )
-            if show_versions_cursor.rowcount is None:
-                raise SnowflakeSQLExecutionError(show_versions_query)
-
-            if show_versions_cursor.rowcount > 0:
-                # allow dropping a package with versions when --force is set
-                if not auto_yes:
-                    raise CouldNotDropApplicationPackageWithVersions(
-                        "Drop versions first, or use --force to override."
-                    )
-
-        # 4. Check distribution of the existing application package
-        actual_distribution = self.get_app_pkg_distribution_in_snowflake
-        if not self.verify_project_distribution(actual_distribution):
-            cc.warning(
-                f"Continuing to execute `snow app teardown` on application package {self.package_name} with distribution '{actual_distribution}'."
-            )
-
-        # 5. If distribution is internal, check if created by the Snowflake CLI
-        row_comment = show_obj_row[COMMENT_COL]
-        if actual_distribution == INTERNAL_DISTRIBUTION:
-            if row_comment in ALLOWED_SPECIAL_COMMENTS:
-                needs_confirm = False
-            else:
-                if needs_confirmation(needs_confirm, auto_yes):
-                    cc.warning(
-                        f"Application package {self.package_name} was not created by Snowflake CLI."
-                    )
-        else:
-            if needs_confirmation(needs_confirm, auto_yes):
-                cc.warning(
-                    f"Application package {self.package_name} in your Snowflake account has distribution property '{EXTERNAL_DISTRIBUTION}' and could be associated with one or more of your listings on Snowflake Marketplace."
-                )
-
-        if needs_confirmation(needs_confirm, auto_yes):
-            should_drop_object = typer.confirm(
-                dedent(
-                    f"""\
-                        Application package details:
-                        Name: {self.app_name}
-                        Created on: {show_obj_row["created_on"]}
-                        Distribution: {actual_distribution}
-                        Owner: {show_obj_row[OWNER_COL]}
-                        Comment: {show_obj_row[COMMENT_COL]}
-                        Are you sure you want to drop it?
-                    """
-                )
-            )
-            if not should_drop_object:
-                cc.message(f"Did not drop application package {self.package_name}.")
-                return  # The user desires to keep the application package, therefore exit gracefully
-
-        # All validations have passed, drop object
-        self.drop_generic_object(
-            object_type="application package",
-            object_name=self.package_name,
-            role=self.package_role,
+        return ApplicationPackageEntity.drop(
+            console=cc,
+            package_name=self.package_name,
+            package_role=self.package_role,
+            force_drop=auto_yes,
         )
-        return  # The application package was successfully dropped, therefore exit gracefully
 
     def process(
         self,

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

@@ -22,7 +22,10 @@
 import typer
 import yaml
 from snowflake.cli._plugins.nativeapp.artifacts import BundleMap
-from snowflake.cli._plugins.nativeapp.common_flags import ValidateOption
+from snowflake.cli._plugins.nativeapp.common_flags import (
+    ForceOption,
+    ValidateOption,
+)
 from snowflake.cli._plugins.workspace.manager import WorkspaceManager
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.commands.decorators import with_project_definition
@@ -162,3 +165,30 @@ def deploy(
         validate=validate,
     )
     return MessageResult("Deployed successfully.")
+
+
+@ws.command(requires_connection=True)
+@with_project_definition()
+def drop(
+    entity_id: str = typer.Option(
+        help=f"""The ID of the entity you want to drop.""",
+    ),
+    # TODO The following options should be generated automatically, depending on the specified entity type
+    force: Optional[bool] = ForceOption,
+    **options,
+):
+    """
+    Drops the specified entity.
+    """
+
+    cli_context = get_cli_context()
+    ws = WorkspaceManager(
+        project_definition=cli_context.project_definition,
+        project_root=cli_context.project_root,
+    )
+
+    ws.perform_action(
+        entity_id,
+        EntityActions.DROP,
+        force_drop=force,
+    )

src/snowflake/cli/api/entities/application_package_entity.py

@@ -4,27 +4,35 @@
 from textwrap import dedent
 from typing import Callable, List, Optional
 
+import typer
 from click import ClickException
 from snowflake.cli._plugins.nativeapp.artifacts import build_bundle
 from snowflake.cli._plugins.nativeapp.bundle_context import BundleContext
 from snowflake.cli._plugins.nativeapp.codegen.compiler import NativeAppCompiler
 from snowflake.cli._plugins.nativeapp.constants import (
     ALLOWED_SPECIAL_COMMENTS,
     COMMENT_COL,
+    EXTERNAL_DISTRIBUTION,
     INTERNAL_DISTRIBUTION,
     NAME_COL,
+    OWNER_COL,
     SPECIAL_COMMENT,
 )
 from snowflake.cli._plugins.nativeapp.exceptions import (
     ApplicationPackageAlreadyExistsError,
     ApplicationPackageDoesNotExistError,
+    CouldNotDropApplicationPackageWithVersions,
     SetupScriptFailedValidation,
 )
+from snowflake.cli._plugins.nativeapp.utils import (
+    needs_confirmation,
+)
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli._plugins.workspace.action_context import ActionContext
 from snowflake.cli.api.console.abc import AbstractConsole
 from snowflake.cli.api.entities.common import EntityBase, get_sql_executor
 from snowflake.cli.api.entities.utils import (
+    drop_generic_object,
     ensure_correct_owner,
     execute_post_deploy_hooks,
     generic_sql_error_handler,
@@ -45,6 +53,7 @@
     get_basic_jinja_env,
 )
 from snowflake.connector import ProgrammingError
+from snowflake.connector.cursor import DictCursor
 
 
 class ApplicationPackageEntity(EntityBase[ApplicationPackageEntityModel]):
@@ -133,6 +142,25 @@ def action_deploy(
                 deploy_to_scratch_stage_fn=lambda *args: None,
             )
 
+    def action_drop(
+        self,
+        ctx: ActionContext,
+        force_drop: bool,
+    ):
+        model = self._entity_model
+        package_name = model.fqn.identifier
+        if model.meta and model.meta.role:
+            package_role = model.meta.role
+        else:
+            package_role = ctx.default_role
+
+        self.drop(
+            console=ctx.console,
+            package_name=package_name,
+            package_role=package_role,
+            force_drop=force_drop,
+        )
+
     @staticmethod
     def get_existing_app_pkg_info(
         package_name: str,
@@ -425,3 +453,101 @@ def get_validation_result(
                     sql_executor.execute_query(
                         f"drop stage if exists {scratch_stage_fqn}"
                     )
+
+    @classmethod
+    def drop(
+        cls,
+        console: AbstractConsole,
+        package_name: str,
+        package_role: str,
+        force_drop: bool,
+    ):
+        sql_executor = get_sql_executor()
+        needs_confirm = True
+
+        # 1. If existing application package is not found, exit gracefully
+        show_obj_row = cls.get_existing_app_pkg_info(
+            package_name=package_name,
+            package_role=package_role,
+        )
+        if show_obj_row is None:
+            console.warning(
+                f"Role {package_role} does not own any application package with the name {package_name}, or the application package does not exist."
+            )
+            return
+
+        # 2. Check for the right owner
+        ensure_correct_owner(row=show_obj_row, role=package_role, obj_name=package_name)
+
+        with sql_executor.use_role(package_role):
+            # 3. Check for versions in the application package
+            show_versions_query = f"show versions in application package {package_name}"
+            show_versions_cursor = sql_executor.execute_query(
+                show_versions_query, cursor_class=DictCursor
+            )
+            if show_versions_cursor.rowcount is None:
+                raise SnowflakeSQLExecutionError(show_versions_query)
+
+            if show_versions_cursor.rowcount > 0:
+                # allow dropping a package with versions when --force is set
+                if not force_drop:
+                    raise CouldNotDropApplicationPackageWithVersions(
+                        "Drop versions first, or use --force to override."
+                    )
+
+        # 4. Check distribution of the existing application package
+        actual_distribution = cls.get_app_pkg_distribution_in_snowflake(
+            package_name=package_name,
+            package_role=package_role,
+        )
+        if not cls.verify_project_distribution(
+            console=console,
+            package_name=package_name,
+            package_role=package_role,
+            package_distribution=actual_distribution,
+        ):
+            console.warning(
+                f"Dropping application package {package_name} with distribution '{actual_distribution}'."
+            )
+
+        # 5. If distribution is internal, check if created by the Snowflake CLI
+        row_comment = show_obj_row[COMMENT_COL]
+        if actual_distribution == INTERNAL_DISTRIBUTION:
+            if row_comment in ALLOWED_SPECIAL_COMMENTS:
+                needs_confirm = False
+            else:
+                if needs_confirmation(needs_confirm, force_drop):
+                    console.warning(
+                        f"Application package {package_name} was not created by Snowflake CLI."
+                    )
+        else:
+            if needs_confirmation(needs_confirm, force_drop):
+                console.warning(
+                    f"Application package {package_name} in your Snowflake account has distribution property '{EXTERNAL_DISTRIBUTION}' and could be associated with one or more of your listings on Snowflake Marketplace."
+                )
+
+        if needs_confirmation(needs_confirm, force_drop):
+            should_drop_object = typer.confirm(
+                dedent(
+                    f"""\
+                        Application package details:
+                        Name: {package_name}
+                        Created on: {show_obj_row["created_on"]}
+                        Distribution: {actual_distribution}
+                        Owner: {show_obj_row[OWNER_COL]}
+                        Comment: {show_obj_row[COMMENT_COL]}
+                        Are you sure you want to drop it?
+                    """
+                )
+            )
+            if not should_drop_object:
+                console.message(f"Did not drop application package {package_name}.")
+                return  # The user desires to keep the application package, therefore exit gracefully
+
+        # All validations have passed, drop object
+        drop_generic_object(
+            console=console,
+            object_type="application package",
+            object_name=package_name,
+            role=package_role,
+        )

src/snowflake/cli/api/entities/common.py

@@ -8,6 +8,7 @@
 class EntityActions(str, Enum):
     BUNDLE = "action_bundle"
     DEPLOY = "action_deploy"
+    DROP = "action_drop"
 
 
 T = TypeVar("T")