skip callers rights

src/snowcli/cli/appify/metadata.py

@@ -11,6 +11,7 @@
 from snowcli.cli.common.sql_execution import SqlExecutionMixin
 from snowcli.cli.object.stage.manager import StageManager
 from snowcli.cli.project.util import to_identifier, to_string_literal
+from snowcli.cli.appify.util import find_row
 
 log = logging.getLogger(__name__)
 
@@ -40,6 +41,11 @@
 ARGUMENTS_REGEX = "(.+) RETURN.+"
 
 
+class ObjectNotFoundError(ClickException):
+    def __init__(self, identifier: str):
+        super().__init__(f"Object did not exist or no rights: {identifier}")
+
+
 class UnexpectedArgumentsFormatError(ClickException):
     def __init__(self, arguments: str):
         super().__init__(f"Unexpected arguments literal: {arguments}")
@@ -81,6 +87,15 @@ def _object_id(self, schema: str, object: str) -> str:
     def _object_literal(self, schema: str, object: str) -> str:
         return to_string_literal(f"{self.database}.{schema}.{object}")
 
+    def _is_callable_callers_rights(self, domain: str, identifier: str) -> str:
+        cursor = self._execute_query(
+            f"describe {domain} {identifier}", cursor_class=DictCursor
+        )
+        execute_as = find_row(cursor, lambda r: r["property"] == "execute as")
+        if not execute_as:
+            raise ObjectNotFoundError(identifier)
+        return execute_as["value"] == "CALLER"
+
     def execute(self) -> None:
         """
         Connects to the target database and dumps metadata into the target path.
@@ -108,6 +123,16 @@ def process_schema(self, schema: str) -> None:
             )
             for object in objects.fetchall():
                 object_name = name_from_object_row(object)
+                # FIXME: need to refactor to split name + arguments so we can quote only the name
+                object_identifier = f"{self._schema_id(schema)}.{object_name}"
+
+                # callers' rights procedures cannot become part of a Native Application
+                if domain == "procedure" and self._is_callable_callers_rights(
+                    domain, object_identifier
+                ):
+                    log.info(f"Skipping callers' rights procedure {object_identifier}")
+                    pass
+
                 literal = self._object_literal(schema, object_name)
                 ddl_cursor = self._execute_query(
                     f"select get_ddl('{domain}', {literal})"

src/snowcli/cli/appify/util.py

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