Skip to content

Commit

Permalink
Workspaces application - drop action (#1572)
Browse files Browse the repository at this point in the history 
* app drop action

* update unit tests

* integration tests
  • Loading branch information
@sfc-gh-gbloom
sfc-gh-gbloom authored Sep 13, 2024
1 parent 869a48e commit 636402c
Show file tree
Hide file tree
Showing 7 changed files with 352 additions and 233 deletions.
34 changes: 10 additions & 24 deletions src/snowflake/cli/_plugins/nativeapp/manager.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from functools import cached_property
from pathlib import Path
from textwrap import dedent
from typing import Generator, List, Optional, TypedDict
from typing import Generator, List, Optional

from snowflake.cli._plugins.connection.util import make_snowsight_url
from snowflake.cli._plugins.nativeapp.artifacts import (
Expand All @@ -38,6 +38,7 @@
from snowflake.cli.api.console import cli_console as cc
from snowflake.cli.api.entities.application_entity import (
ApplicationEntity,
ApplicationOwnedObject,
)
from snowflake.cli.api.entities.application_package_entity import (
ApplicationPackageEntity,
Expand All @@ -57,8 +58,6 @@
from snowflake.cli.api.sql_execution import SqlExecutionMixin
from snowflake.connector import DictCursor, ProgrammingError

ApplicationOwnedObject = TypedDict("ApplicationOwnedObject", {"name": str, "type": str})


class NativeAppCommandProcessor(ABC):
@abstractmethod
Expand Down Expand Up @@ -246,32 +245,19 @@ def get_existing_app_pkg_info(self) -> Optional[dict]:
package_role=self.package_role,
)

def get_objects_owned_by_application(self) -> List[ApplicationOwnedObject]:
"""
Returns all application objects owned by this application.
"""
with self.use_role(self.app_role):
results = self._execute_query(
f"show objects owned by application {self.app_name}"
).fetchall()
return [{"name": row[1], "type": row[2]} for row in results]
def get_objects_owned_by_application(self):
return ApplicationEntity.get_objects_owned_by_application(
app_name=self.app_name,
app_role=self.app_role,
)

def _application_objects_to_str(
self, application_objects: list[ApplicationOwnedObject]
) -> str:
"""
Returns a list in an "(Object Type) Object Name" format. Database-level and schema-level object names are fully qualified:
(COMPUTE_POOL) POOL_NAME
(DATABASE) DB_NAME
(SCHEMA) DB_NAME.PUBLIC
...
"""
return "\n".join(
[self._application_object_to_str(obj) for obj in application_objects]
)
return ApplicationEntity.application_objects_to_str(application_objects)

def _application_object_to_str(self, obj: ApplicationOwnedObject) -> str:
return f"({obj['type']}) {obj['name']}"
def _application_object_to_str(self, obj: ApplicationOwnedObject):
return ApplicationEntity.application_object_to_str(obj)

def get_snowsight_url(self) -> str:
"""Returns the URL that can be used to visit this app via Snowsight."""
Expand Down
161 changes: 9 additions & 152 deletions src/snowflake/cli/_plugins/nativeapp/teardown_processor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,179 +15,36 @@
from __future__ import annotations

from pathlib import Path
from textwrap import dedent
from typing import Dict, Optional

import typer
from snowflake.cli._plugins.nativeapp.constants import (
ALLOWED_SPECIAL_COMMENTS,
COMMENT_COL,
OWNER_COL,
)
from snowflake.cli._plugins.nativeapp.manager import (
NativeAppCommandProcessor,
NativeAppManager,
)
from snowflake.cli._plugins.nativeapp.utils import (
needs_confirmation,
)
from snowflake.cli.api.console import cli_console as cc
from snowflake.cli.api.entities.application_entity import (
ApplicationEntity,
)
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.connector import ProgrammingError


class NativeAppTeardownProcessor(NativeAppManager, NativeAppCommandProcessor):
def __init__(self, project_definition: Dict, project_root: Path):
super().__init__(project_definition, project_root)

def drop_generic_object(
self, object_type: str, object_name: str, role: str, cascade: bool = False
):
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
):
"""
Attempts to drop the application object if all validations and user prompts allow so.
"""

needs_confirm = True

# 1. If existing application is not found, exit gracefully
show_obj_row = self.get_existing_app_info()
if show_obj_row is None:
cc.warning(
f"Role {self.app_role} does not own any application object with the name {self.app_name}, or the application object does not exist."
)
return

# 2. Check for the right owner
ensure_correct_owner(
row=show_obj_row, role=self.app_role, obj_name=self.app_name
)

# 3. Check if created by the Snowflake CLI
row_comment = show_obj_row[COMMENT_COL]
if row_comment not in ALLOWED_SPECIAL_COMMENTS and needs_confirmation(
needs_confirm, auto_yes
):
should_drop_object = typer.confirm(
dedent(
f"""\
Application object {self.app_name} was not created by Snowflake CLI.
Application object details:
Name: {self.app_name}
Created on: {show_obj_row["created_on"]}
Source: {show_obj_row["source"]}
Owner: {show_obj_row[OWNER_COL]}
Comment: {show_obj_row[COMMENT_COL]}
Version: {show_obj_row["version"]}
Patch: {show_obj_row["patch"]}
Are you sure you want to drop it?
"""
)
)
if not should_drop_object:
cc.message(f"Did not drop application object {self.app_name}.")
# The user desires to keep the app, therefore we can't proceed since it would
# leave behind an orphan app when we get to dropping the package
raise typer.Abort()

# 4. Check for application objects owned by the application
# This query will fail if the application package has already been dropped, so handle this case gracefully
has_objects_to_drop = False
message_prefix = ""
cascade_true_message = ""
cascade_false_message = ""
interactive_prompt = ""
non_interactive_abort = ""
try:
if application_objects := self.get_objects_owned_by_application():
has_objects_to_drop = True
message_prefix = (
f"The following objects are owned by application {self.app_name}"
)
cascade_true_message = f"{message_prefix} and will be dropped:"
cascade_false_message = f"{message_prefix} and will NOT be dropped:"
interactive_prompt = "Would you like to drop these objects in addition to the application? [y/n/ABORT]"
non_interactive_abort = "Re-run teardown again with --cascade or --no-cascade to specify whether these objects should be dropped along with the application"
except ProgrammingError as e:
if e.errno != APPLICATION_NO_LONGER_AVAILABLE:
raise
application_objects = []
message_prefix = f"Could not determine which objects are owned by application {self.app_name}"
has_objects_to_drop = True # potentially, but we don't know what they are
cascade_true_message = (
f"{message_prefix}, an unknown number of objects will be dropped."
)
cascade_false_message = f"{message_prefix}, they will NOT be dropped."
interactive_prompt = f"Would you like to drop an unknown set of objects in addition to the application? [y/n/ABORT]"
non_interactive_abort = f"Re-run teardown again with --cascade or --no-cascade to specify whether any objects should be dropped along with the application."

if has_objects_to_drop:
if cascade is True:
# If the user explicitly passed the --cascade flag
cc.message(cascade_true_message)
with cc.indented():
for obj in application_objects:
cc.message(self._application_object_to_str(obj))
elif cascade is False:
# If the user explicitly passed the --no-cascade flag
cc.message(cascade_false_message)
with cc.indented():
for obj in application_objects:
cc.message(self._application_object_to_str(obj))
elif interactive:
# If the user didn't pass any cascade flag and the session is interactive
cc.message(message_prefix)
with cc.indented():
for obj in application_objects:
cc.message(self._application_object_to_str(obj))
user_response = typer.prompt(
interactive_prompt,
show_default=False,
default="ABORT",
).lower()
if user_response in ["y", "yes"]:
cascade = True
elif user_response in ["n", "no"]:
cascade = False
else:
raise typer.Abort()
else:
# Else abort since we don't know what to do and can't ask the user
cc.message(message_prefix)
with cc.indented():
for obj in application_objects:
cc.message(self._application_object_to_str(obj))
cc.message(non_interactive_abort)
raise typer.Abort()
elif cascade is None:
# If there's nothing to drop, set cascade to an explicit False value
cascade = False

# 5. All validations have passed, drop object
self.drop_generic_object(
object_type="application",
object_name=self.app_name,
role=self.app_role,
return ApplicationEntity.drop(
console=cc,
app_name=self.app_name,
app_role=self.app_role,
auto_yes=auto_yes,
interactive=interactive,
cascade=cascade,
)
return # The application object was successfully dropped, therefore exit gracefully

def drop_package(self, auto_yes: bool):
return ApplicationPackageEntity.drop(
Expand Down
8 changes: 8 additions & 0 deletions src/snowflake/cli/_plugins/workspace/commands.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,13 @@ def drop(
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
interactive: bool = InteractiveOption,
force: Optional[bool] = ForceOption,
cascade: Optional[bool] = typer.Option(
None,
help=f"""Whether to drop all application objects owned by the application within the account. Default: false.""",
show_default=False,
),
**options,
):
"""
Expand All @@ -194,6 +200,8 @@ def drop(
entity_id,
EntityActions.DROP,
force_drop=force,
interactive=interactive,
cascade=cascade,
)


Expand Down
Loading

0 comments on commit 636402c

Please sign in to comment.