Adding app deploy command

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

@@ -224,3 +224,22 @@ def app_teardown(
     )
     processor.process(force)
     return MessageResult(f"Teardown is now complete.")
+
+
+@app.command("deploy", requires_connection=True)
+@with_project_definition("native_app")
+def app_deploy(
+    **options,
+) -> CommandResult:
+    """
+    Syncs the local changes to the stage without creating or updating the application.
+    """
+    manager = NativeAppManager(
+        project_definition=cli_context.project_definition,
+        project_root=cli_context.project_root,
+    )
+
+    manager.build_bundle()
+    manager.process()
+
+    return MessageResult(f"Deployed successfully.")

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

@@ -6,6 +6,7 @@
 from textwrap import dedent
 from typing import List, Optional
 
+import jinja2
 from snowflake.cli.api.console import cli_console as cc
 from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
 from snowflake.cli.api.project.definition import (
@@ -27,12 +28,21 @@
     translate_artifact,
 )
 from snowflake.cli.plugins.nativeapp.constants import (
+    ALLOWED_SPECIAL_COMMENTS,
+    COMMENT_COL,
     ERROR_MESSAGE_606,
     ERROR_MESSAGE_2043,
+    INTERNAL_DISTRIBUTION,
     NAME_COL,
     OWNER_COL,
+    SPECIAL_COMMENT,
+)
+from snowflake.cli.plugins.nativeapp.exceptions import (
+    ApplicationPackageAlreadyExistsError,
+    InvalidPackageScriptError,
+    MissingPackageScriptError,
+    UnexpectedOwnerError,
 )
-from snowflake.cli.plugins.nativeapp.exceptions import UnexpectedOwnerError
 from snowflake.cli.plugins.object.stage.diff import (
     DiffResult,
     stage_diff,
@@ -327,3 +337,101 @@ 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)
         return make_snowsight_url(self._conn, f"/#/apps/application/{name}")
+
+    def create_app_package(self) -> None:
+        """
+        Creates the application package with our up-to-date stage if none exists.
+        """
+
+        # 1. Check for existing existing application package
+        show_obj_row = self.get_existing_app_pkg_info()
+
+        if show_obj_row:
+            # 1. Check for the right owner role
+            ensure_correct_owner(
+                row=show_obj_row, role=self.package_role, obj_name=self.package_name
+            )
+
+            # 2. 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 run` on application package {self.package_name} with distribution '{actual_distribution}'."
+                )
+
+            # 3. If actual_distribution is external, skip comment check
+            if actual_distribution == INTERNAL_DISTRIBUTION:
+                row_comment = show_obj_row[COMMENT_COL]
+
+                if row_comment not in ALLOWED_SPECIAL_COMMENTS:
+                    raise ApplicationPackageAlreadyExistsError(self.package_name)
+
+            return
+
+        # If no application package pre-exists, create an application package, with the specified distribution in the project definition file.
+        with self.use_role(self.package_role):
+            cc.step(f"Creating new application package {self.package_name} in account.")
+            self._execute_query(
+                dedent(
+                    f"""\
+                    create application package {self.package_name}
+                        comment = {SPECIAL_COMMENT}
+                        distribution = {self.package_distribution}
+                """
+                )
+            )
+
+    def _apply_package_scripts(self) -> None:
+        """
+        Assuming the application package exists and we are using the correct role,
+        applies all package scripts in-order to the application package.
+        """
+        env = jinja2.Environment(
+            loader=jinja2.loaders.FileSystemLoader(self.project_root),
+            keep_trailing_newline=True,
+            undefined=jinja2.StrictUndefined,
+        )
+
+        queued_queries = []
+        for relpath in self.package_scripts:
+            try:
+                template = env.get_template(relpath)
+                result = template.render(dict(package_name=self.package_name))
+                queued_queries.append(result)
+
+            except jinja2.TemplateNotFound as e:
+                raise MissingPackageScriptError(e.name)
+
+            except jinja2.TemplateSyntaxError as e:
+                raise InvalidPackageScriptError(e.name, e)
+
+            except jinja2.UndefinedError as e:
+                raise InvalidPackageScriptError(relpath, e)
+
+        # once we're sure all the templates expanded correctly, execute all of them
+        try:
+            if self.package_warehouse:
+                self._execute_query(f"use warehouse {self.package_warehouse}")
+
+            for i, queries in enumerate(queued_queries):
+                cc.step(f"Applying package script: {self.package_scripts[i]}")
+                self._execute_queries(queries)
+        except ProgrammingError as err:
+            generic_sql_error_handler(
+                err, role=self.package_role, warehouse=self.package_warehouse
+            )
+
+    def deploy(self) -> DiffResult:
+        """app deploy process"""
+
+        # 1. Create an empty application package, if none exists
+        self.create_app_package()
+
+        with self.use_role(self.package_role):
+            # 2. now that the application package exists, create shared data
+            self._apply_package_scripts()
+
+            # 3. Upload files from deploy root local folder to the above stage
+            diff = self.sync_deploy_root_with_stage(self.package_role)
+
+        return diff

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

@@ -2,26 +2,17 @@
 from textwrap import dedent
 from typing import Optional
 
-import jinja2
 import typer
 from click import UsageError
 from snowflake.cli.api.console import cli_console as cc
 from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
 from snowflake.cli.api.project.schemas.native_app.native_app import NativeApp
 from snowflake.cli.plugins.nativeapp.constants import (
-    ALLOWED_SPECIAL_COMMENTS,
-    COMMENT_COL,
-    INTERNAL_DISTRIBUTION,
-    LOOSE_FILES_MAGIC_VERSION,
     SPECIAL_COMMENT,
     VERSION_COL,
 )
 from snowflake.cli.plugins.nativeapp.exceptions import (
-    ApplicationAlreadyExistsError,
-    ApplicationPackageAlreadyExistsError,
     ApplicationPackageDoesNotExistError,
-    InvalidPackageScriptError,
-    MissingPackageScriptError,
 )
 from snowflake.cli.plugins.nativeapp.manager import (
     NativeAppCommandProcessor,
@@ -30,8 +21,6 @@
     generic_sql_error_handler,
 )
 from snowflake.cli.plugins.nativeapp.policy import PolicyBase
-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 SnowflakeCursor
 
@@ -42,173 +31,6 @@ class NativeAppRunProcessor(NativeAppManager, NativeAppCommandProcessor):
     def __init__(self, project_definition: NativeApp, project_root: Path):
         super().__init__(project_definition, project_root)
 
-    def create_app_package(self) -> None:
-        """
-        Creates the application package with our up-to-date stage if none exists.
-        """
-
-        # 1. Check for existing existing application package
-        show_obj_row = self.get_existing_app_pkg_info()
-
-        if show_obj_row:
-            # 1. Check for the right owner role
-            ensure_correct_owner(
-                row=show_obj_row, role=self.package_role, obj_name=self.package_name
-            )
-
-            # 2. 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 run` on application package {self.package_name} with distribution '{actual_distribution}'."
-                )
-
-            # 3. If actual_distribution is external, skip comment check
-            if actual_distribution == INTERNAL_DISTRIBUTION:
-                row_comment = show_obj_row[COMMENT_COL]
-
-                if row_comment not in ALLOWED_SPECIAL_COMMENTS:
-                    raise ApplicationPackageAlreadyExistsError(self.package_name)
-
-            return
-
-        # If no application package pre-exists, create an application package, with the specified distribution in the project definition file.
-        with self.use_role(self.package_role):
-            cc.step(f"Creating new application package {self.package_name} in account.")
-            self._execute_query(
-                dedent(
-                    f"""\
-                    create application package {self.package_name}
-                        comment = {SPECIAL_COMMENT}
-                        distribution = {self.package_distribution}
-                """
-                )
-            )
-
-    def _apply_package_scripts(self) -> None:
-        """
-        Assuming the application package exists and we are using the correct role,
-        applies all package scripts in-order to the application package.
-        """
-        env = jinja2.Environment(
-            loader=jinja2.loaders.FileSystemLoader(self.project_root),
-            keep_trailing_newline=True,
-            undefined=jinja2.StrictUndefined,
-        )
-
-        queued_queries = []
-        for relpath in self.package_scripts:
-            try:
-                template = env.get_template(relpath)
-                result = template.render(dict(package_name=self.package_name))
-                queued_queries.append(result)
-
-            except jinja2.TemplateNotFound as e:
-                raise MissingPackageScriptError(e.name)
-
-            except jinja2.TemplateSyntaxError as e:
-                raise InvalidPackageScriptError(e.name, e)
-
-            except jinja2.UndefinedError as e:
-                raise InvalidPackageScriptError(relpath, e)
-
-        # once we're sure all the templates expanded correctly, execute all of them
-        try:
-            if self.package_warehouse:
-                self._execute_query(f"use warehouse {self.package_warehouse}")
-
-            for i, queries in enumerate(queued_queries):
-                cc.step(f"Applying package script: {self.package_scripts[i]}")
-                self._execute_queries(queries)
-        except ProgrammingError as err:
-            generic_sql_error_handler(
-                err, role=self.package_role, warehouse=self.package_warehouse
-            )
-
-    def _create_dev_app(self, diff: DiffResult) -> None:
-        """
-        (Re-)creates the application object with our up-to-date stage.
-        """
-        with self.use_role(self.app_role):
-
-            # 1. Need to use a warehouse to create an application object
-            try:
-                if self.application_warehouse:
-                    self._execute_query(f"use warehouse {self.application_warehouse}")
-            except ProgrammingError as err:
-                generic_sql_error_handler(
-                    err=err, role=self.app_role, warehouse=self.application_warehouse
-                )
-
-            # 2. Check for an existing application object by the same name
-            show_app_row = self.get_existing_app_info()
-
-            # 3. If existing application object is found, perform a few validations and upgrade the application object.
-            if show_app_row:
-
-                # Check if not created by Snowflake CLI or not created using "files on a named stage" / stage dev mode.
-                if show_app_row[COMMENT_COL] not in ALLOWED_SPECIAL_COMMENTS or (
-                    show_app_row[VERSION_COL] != LOOSE_FILES_MAGIC_VERSION
-                ):
-                    raise ApplicationAlreadyExistsError(self.app_name)
-
-                # Check for the right owner
-                ensure_correct_owner(
-                    row=show_app_row, role=self.app_role, obj_name=self.app_name
-                )
-
-                # If all the above checks are in order, proceed to upgrade
-                try:
-                    if diff.has_changes():
-                        cc.step(
-                            f"Upgrading existing application object {self.app_name}."
-                        )
-                        self._execute_query(
-                            f"alter application {self.app_name} upgrade using @{self.stage_fqn}"
-                        )
-
-                    # ensure debug_mode is up-to-date
-                    self._execute_query(
-                        f"alter application {self.app_name} set debug_mode = {self.debug_mode}"
-                    )
-
-                    return
-
-                except ProgrammingError as err:
-                    generic_sql_error_handler(err)
-
-            # 4. If no existing application object is found, create an application object using "files on a named stage" / stage dev mode.
-            cc.step(f"Creating new application {self.app_name} in account.")
-
-            if self.app_role != self.package_role:
-                with self.use_role(new_role=self.package_role):
-                    self._execute_queries(
-                        dedent(
-                            f"""\
-                        grant install, develop on application package {self.package_name} to role {self.app_role};
-                        grant usage on schema {self.package_name}.{self.stage_schema} to role {self.app_role};
-                        grant read on stage {self.stage_fqn} to role {self.app_role};
-                        """
-                        )
-                    )
-
-            stage_name = StageManager.quote_stage_name(self.stage_fqn)
-
-            try:
-                self._execute_query(
-                    dedent(
-                        f"""\
-                    create application {self.app_name}
-                        from application package {self.package_name}
-                        using {stage_name}
-                        debug_mode = {self.debug_mode}
-                        comment = {SPECIAL_COMMENT}
-                    """
-                    )
-                )
-            except ProgrammingError as err:
-                generic_sql_error_handler(err)
-
     def get_all_existing_versions(self) -> SnowflakeCursor:
         """
         Get all existing versions, if defined, for an application package.
@@ -385,15 +207,5 @@ def process(
             )
             return
 
-        # 1. Create an empty application package, if none exists
-        self.create_app_package()
-
-        with self.use_role(self.package_role):
-            # 2. now that the application package exists, create shared data
-            self._apply_package_scripts()
-
-            # 3. Upload files from deploy root local folder to the above stage
-            diff = self.sync_deploy_root_with_stage(self.package_role)
-
-        # 4. Create an application if none exists, else upgrade the application
+        diff = self.deploy()
         self._create_dev_app(diff)