Create a pattern for drilling down to a single "show objects like" row

src/snowflake/cli/api/project/util.py

@@ -170,3 +170,11 @@ def escape_like_pattern(pattern: str, escape_sequence: str = r"\\") -> str:
         "_", rf"{escape_sequence}_"
     )
     return pattern
+
+
+def identifier_to_show_like_pattern(identifier: str) -> str:
+    """
+    Takes an identifier and converts it into a pattern to be used with SHOW ... LIKE ... to get all rows with name
+    matching this identifier
+    """
+    return f"'{escape_like_pattern(unquote_identifier(identifier))}'"

src/snowflake/cli/api/sql_execution.py

@@ -5,10 +5,16 @@
 from functools import cached_property
 from io import StringIO
 from textwrap import dedent
-from typing import Iterable
+from typing import Iterable, Optional
 
 from click import ClickException
 from snowflake.cli.api.cli_global_context import cli_context
+from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
+from snowflake.cli.api.project.util import (
+    identifier_to_show_like_pattern,
+    unquote_identifier,
+)
+from snowflake.cli.api.utils.cursor import find_first_row
 from snowflake.connector.cursor import DictCursor, SnowflakeCursor
 from snowflake.connector.errors import ProgrammingError
 
@@ -137,3 +143,27 @@ def to_fully_qualified_name(self, name: str):
 
         schema = self._conn.schema or "public"
         return f"{self._conn.database}.{schema}.{name}".upper()
+
+    def show_specific_object(
+        self,
+        object_type_plural: str,
+        unqualified_name: str,
+        name_col: str = "name",
+        in_clause: str = "",
+    ) -> Optional[dict]:
+        """
+        Executes a "show <objects> like" query for a particular entity with a
+        given (unqualified) name. This command is useful when the corresponding
+        "describe <object>" query does not provide the information you seek.
+        """
+        show_obj_query = f"show {object_type_plural} like {identifier_to_show_like_pattern(unqualified_name)} {in_clause}".strip()
+        show_obj_cursor = self._execute_query(  # type: ignore
+            show_obj_query, cursor_class=DictCursor
+        )
+        if show_obj_cursor.rowcount is None:
+            raise SnowflakeSQLExecutionError(show_obj_query)
+        show_obj_row = find_first_row(
+            show_obj_cursor,
+            lambda row: row[name_col] == unquote_identifier(unqualified_name),
+        )
+        return show_obj_row

src/snowflake/cli/api/utils/cursor.py

@@ -0,0 +1,18 @@
+from typing import Callable, List, Optional
+
+from snowflake.connector.cursor import DictCursor
+
+
+def _rows_generator(cursor: DictCursor, predicate: Callable[[dict], bool]):
+    return (row for row in cursor.fetchall() if predicate(row))
+
+
+def find_all_rows(cursor: DictCursor, predicate: Callable[[dict], bool]) -> List[dict]:
+    return list(_rows_generator(cursor, predicate))
+
+
+def find_first_row(
+    cursor: DictCursor, predicate: Callable[[dict], bool]
+) -> Optional[dict]:
+    """Returns the first row that matches the predicate, or None."""
+    return next(_rows_generator(cursor, predicate), None)

src/snowflake/cli/plugins/nativeapp/manager.py

@@ -32,14 +32,12 @@
     OWNER_COL,
 )
 from snowflake.cli.plugins.nativeapp.exceptions import UnexpectedOwnerError
-from snowflake.cli.plugins.nativeapp.utils import find_first_row
 from snowflake.cli.plugins.object.stage.diff import (
     DiffResult,
     stage_diff,
     sync_local_diff_with_stage,
 )
 from snowflake.connector import ProgrammingError
-from snowflake.connector.cursor import DictCursor
 
 log = logging.getLogger(__name__)
 
@@ -304,23 +302,9 @@ def get_existing_app_info(self) -> Optional[dict]:
         It executes a 'show applications like' query and returns the result as single row, if one exists.
         """
         with self.use_role(self.app_role):
-            show_obj_query = (
-                f"show applications like '{unquote_identifier(self.app_name)}'"
+            return self.show_specific_object(
+                "applications", self.app_name, name_col=NAME_COL
             )
-            show_obj_cursor = self._execute_query(
-                show_obj_query,
-                cursor_class=DictCursor,
-            )
-
-            if show_obj_cursor.rowcount is None:
-                raise SnowflakeSQLExecutionError(show_obj_query)
-
-            show_obj_row = find_first_row(
-                show_obj_cursor,
-                lambda row: row[NAME_COL] == unquote_identifier(self.app_name),
-            )
-
-            return show_obj_row
 
     def get_existing_app_pkg_info(self) -> Optional[dict]:
         """
@@ -329,21 +313,10 @@ def get_existing_app_pkg_info(self) -> Optional[dict]:
         """
 
         with self.use_role(self.package_role):
-            show_obj_query = f"show application packages like '{unquote_identifier(self.package_name)}'"
-            show_obj_cursor = self._execute_query(
-                show_obj_query, cursor_class=DictCursor
+            return self.show_specific_object(
+                "application packages", self.package_name, name_col=NAME_COL
             )
 
-            if show_obj_cursor.rowcount is None:
-                raise SnowflakeSQLExecutionError(show_obj_query)
-
-            show_obj_row = find_first_row(
-                show_obj_cursor,
-                lambda row: row[NAME_COL] == unquote_identifier(self.package_name),
-            )
-
-            return show_obj_row  # Can be None or a dict
-
     def get_snowsight_url(self) -> str:
         """Returns the URL that can be used to visit this app via Snowsight."""
         name = unquote_identifier(self.app_name)

src/snowflake/cli/plugins/nativeapp/run_processor.py

@@ -7,7 +7,6 @@
 from click import UsageError
 from rich import print
 from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
-from snowflake.cli.api.project.util import unquote_identifier
 from snowflake.cli.plugins.nativeapp.constants import (
     COMMENT_COL,
     INTERNAL_DISTRIBUTION,
@@ -29,11 +28,10 @@
     generic_sql_error_handler,
 )
 from snowflake.cli.plugins.nativeapp.policy import PolicyBase
-from snowflake.cli.plugins.nativeapp.utils import find_first_row
 from snowflake.cli.plugins.object.stage.diff import DiffResult
 from snowflake.cli.plugins.object.stage.manager import StageManager
 from snowflake.connector import ProgrammingError
-from snowflake.connector.cursor import DictCursor, SnowflakeCursor
+from snowflake.connector.cursor import SnowflakeCursor
 
 UPGRADE_RESTRICTION_CODES = {93044, 93055, 93045, 93046}
 
@@ -227,27 +225,20 @@ def get_existing_version_info(self, version: str) -> Optional[dict]:
         It executes a 'show versions like ... in application package' query and returns the result as single row, if one exists.
         """
         with self.use_role(self.package_role):
-            show_obj_query = f"show versions like '{unquote_identifier(version)}' in application package {self.package_name}"
-
             try:
-                show_obj_cursor = self._execute_query(
-                    show_obj_query, cursor_class=DictCursor
+                version_obj = self.show_specific_object(
+                    "versions",
+                    version,
+                    name_col=VERSION_COL,
+                    in_clause=f"in application package {self.package_name}",
                 )
             except ProgrammingError as err:
                 if err.msg.__contains__("does not exist or not authorized"):
                     raise ApplicationPackageDoesNotExistError(self.package_name)
                 else:
                     generic_sql_error_handler(err=err, role=self.package_role)
 
-            if show_obj_cursor.rowcount is None:
-                raise SnowflakeSQLExecutionError(show_obj_query)
-
-            show_obj_row = find_first_row(
-                show_obj_cursor,
-                lambda row: row[VERSION_COL] == unquote_identifier(version),
-            )
-
-            return show_obj_row
+            return version_obj
 
     def drop_application_before_upgrade(self, policy: PolicyBase, is_interactive: bool):
         """

src/snowflake/cli/plugins/nativeapp/utils.py

@@ -1,22 +1,4 @@
 from sys import stdin, stdout
-from typing import Callable, List, Optional
-
-from snowflake.connector.cursor import DictCursor
-
-
-def _rows_generator(cursor: DictCursor, predicate: Callable[[dict], bool]):
-    return (row for row in cursor.fetchall() if predicate(row))
-
-
-def find_all_rows(cursor: DictCursor, predicate: Callable[[dict], bool]) -> List[dict]:
-    return list(_rows_generator(cursor, predicate))
-
-
-def find_first_row(
-    cursor: DictCursor, predicate: Callable[[dict], bool]
-) -> Optional[dict]:
-    """Returns the first row that matches the predicate, or None."""
-    return next(_rows_generator(cursor, predicate), None)
 
 
 def needs_confirmation(needs_confirm: bool, auto_yes: bool) -> bool:

src/snowflake/cli/plugins/nativeapp/version/version_processor.py

@@ -8,6 +8,10 @@
 from rich import print
 from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
 from snowflake.cli.api.project.util import unquote_identifier
+from snowflake.cli.api.utils.cursor import (
+    find_all_rows,
+    find_first_row,
+)
 from snowflake.cli.plugins.nativeapp.artifacts import find_version_info_in_manifest_file
 from snowflake.cli.plugins.nativeapp.constants import VERSION_COL
 from snowflake.cli.plugins.nativeapp.exceptions import (
@@ -20,10 +24,6 @@
 )
 from snowflake.cli.plugins.nativeapp.policy import PolicyBase
 from snowflake.cli.plugins.nativeapp.run_processor import NativeAppRunProcessor
-from snowflake.cli.plugins.nativeapp.utils import (
-    find_all_rows,
-    find_first_row,
-)
 from snowflake.connector import ProgrammingError
 from snowflake.connector.cursor import DictCursor
 

src/snowflake/cli/plugins/spcs/image_repository/manager.py

@@ -1,15 +1,12 @@
-from typing import Dict
 from urllib.parse import urlparse
 
 from click import ClickException
 from snowflake.cli.api.constants import ObjectType
 from snowflake.cli.api.project.util import (
-    escape_like_pattern,
     is_valid_unquoted_identifier,
 )
 from snowflake.cli.api.sql_execution import SqlExecutionMixin
 from snowflake.cli.plugins.spcs.common import handle_object_already_exists
-from snowflake.connector.cursor import DictCursor
 from snowflake.connector.errors import ProgrammingError
 
 
@@ -23,40 +20,16 @@ def get_schema(self):
     def get_role(self):
         return self._conn.role
 
-    def get_repository_row(self, repo_name: str) -> Dict:
+    def get_repository_url(self, repo_name: str, with_scheme: bool = True):
         if not is_valid_unquoted_identifier(repo_name):
             raise ValueError(
                 f"repo_name '{repo_name}' is not a valid unquoted Snowflake identifier"
             )
-
-        repo_name = repo_name.upper()
-
-        # because image repositories only support unquoted identifiers, SHOW LIKE should only return one or zero rows
-        repository_list_query = (
-            f"show image repositories like '{escape_like_pattern(repo_name)}'"
-        )
-
-        result_set = self._execute_schema_query(
-            repository_list_query, cursor_class=DictCursor
-        )
-        results = result_set.fetchall()
-
-        if len(results) == 0:
+        repo_row = self.show_specific_object("image repositories", repo_name)
+        if repo_row is None:
             raise ClickException(
                 f"Image repository '{repo_name}' does not exist in database '{self.get_database()}' and schema '{self.get_schema()}' or not authorized."
             )
-        elif len(results) > 1:
-            raise ClickException(
-                f"Found more than one image repository with name matching '{repo_name}'. This is unexpected."
-            )
-        return results[0]
-
-    def get_repository_url(self, repo_name: str, with_scheme: bool = True):
-        if not is_valid_unquoted_identifier(repo_name):
-            raise ValueError(
-                f"repo_name '{repo_name}' is not a valid unquoted Snowflake identifier"
-            )
-        repo_row = self.get_repository_row(repo_name)
         if with_scheme:
             return f"https://{repo_row['repository_url']}"
         else:

tests/nativeapp/test_manager.py

@@ -1,6 +1,4 @@
-import unittest
 from textwrap import dedent
-
 from snowflake.cli.plugins.nativeapp.constants import (
     LOOSE_FILES_MAGIC_VERSION,
     NAME_COL,
@@ -14,7 +12,6 @@
 )
 from snowflake.cli.plugins.object.stage.diff import DiffResult
 from snowflake.cli.api.project.definition_manager import DefinitionManager
-from snowflake.connector import ProgrammingError
 from snowflake.connector.cursor import DictCursor
 
 from tests.nativeapp.patch_utils import (
@@ -415,7 +412,8 @@ def test_get_existing_app_pkg_info_app_pkg_exists(mock_execute, temp_dir, mock_c
                     [],
                 ),
                 mock.call(
-                    "show application packages like 'APP_PKG'", cursor_class=DictCursor
+                    r"show application packages like 'APP\\_PKG'",
+                    cursor_class=DictCursor,
                 ),
             ),
             (None, mock.call("use role old_role")),
@@ -451,7 +449,8 @@ def test_get_existing_app_pkg_info_app_pkg_does_not_exist(
             (
                 mock_cursor([], []),
                 mock.call(
-                    "show application packages like 'APP_PKG'", cursor_class=DictCursor
+                    r"show application packages like 'APP\\_PKG'",
+                    cursor_class=DictCursor,
                 ),
             ),
             (None, mock.call("use role old_role")),