Add release channels add-accounts remove-accounts commands

RELEASE-NOTES.md

@@ -28,6 +28,7 @@
 * `snow app version create` now returns version, patch, and label in JSON format.
 * Add ability to specify release channel when creating application instance from release directive: `snow app run --from-release-directive --channel=<channel>`
 * Add ability to list release channels through `snow app release-channel list` command
+* Add ability to add and remove accounts from release channels through `snow app release-channel add-accounts` and snow app release-channel remove-accounts` commands.
 
 ## Fixes and improvements
 * Fixed crashes with older x86_64 Intel CPUs.

src/snowflake/cli/_plugins/nativeapp/entities/application.py

@@ -26,7 +26,6 @@
 from snowflake.cli._plugins.nativeapp.constants import (
     ALLOWED_SPECIAL_COMMENTS,
     COMMENT_COL,
-    DEFAULT_CHANNEL,
     OWNER_COL,
 )
 from snowflake.cli._plugins.nativeapp.entities.application_package import (
@@ -86,8 +85,6 @@
     append_test_resource_suffix,
     extract_schema,
     identifier_for_url,
-    identifier_in_list,
-    same_identifiers,
     to_identifier,
     unquote_identifier,
 )
@@ -360,8 +357,8 @@ def action_deploy(
 
         # same-account release directive
         if from_release_directive:
-            release_channel = _get_verified_release_channel(
-                package_entity, release_channel
+            release_channel = package_entity.get_sanitized_release_channel(
+                release_channel
             )
 
             self.create_or_upgrade_app(
@@ -1025,28 +1022,3 @@ def _application_objects_to_str(
 
 def _application_object_to_str(obj: ApplicationOwnedObject) -> str:
     return f"({obj['type']}) {obj['name']}"
-
-
-def _get_verified_release_channel(
-    package_entity: ApplicationPackageEntity,
-    release_channel: Optional[str],
-) -> Optional[str]:
-    release_channel = release_channel or DEFAULT_CHANNEL
-    available_release_channels = get_snowflake_facade().show_release_channels(
-        package_entity.name, role=package_entity.role
-    )
-    if available_release_channels:
-        release_channel_names = [c["name"] for c in available_release_channels]
-        if not identifier_in_list(release_channel, release_channel_names):
-            raise UsageError(
-                f"Release channel '{release_channel}' is not available for application package {package_entity.name}. Available release channels: ({', '.join(release_channel_names)})."
-            )
-    else:
-        if same_identifiers(release_channel, DEFAULT_CHANNEL):
-            return None
-        else:
-            raise UsageError(
-                f"Release channels are not enabled for application package {package_entity.name}."
-            )
-
-    return release_channel

src/snowflake/cli/_plugins/nativeapp/entities/application_package.py

@@ -8,7 +8,7 @@
 from typing import Any, List, Literal, Optional, Set, Union
 
 import typer
-from click import BadOptionUsage, ClickException
+from click import BadOptionUsage, ClickException, UsageError
 from pydantic import Field, field_validator
 from snowflake.cli._plugins.connection.util import UIParameter
 from snowflake.cli._plugins.nativeapp.artifacts import (
@@ -97,7 +97,6 @@
     VALID_IDENTIFIER_REGEX,
     append_test_resource_suffix,
     extract_schema,
-    identifier_in_list,
     identifier_to_show_like_pattern,
     same_identifiers,
     sql_match,
@@ -622,6 +621,69 @@ def action_version_drop(
             f"Version {version} in application package {self.name} dropped successfully."
         )
 
+    def _validate_target_accounts(self, accounts: list[str]) -> None:
+        """
+        Validates the target accounts provided by the user.
+        """
+        for account in accounts:
+            if not re.fullmatch(
+                f"{VALID_IDENTIFIER_REGEX}\\.{VALID_IDENTIFIER_REGEX}", account
+            ):
+                raise ClickException(
+                    f"Target account {account} is not in a valid format. Make sure you provide the target account in the format 'org.account'."
+                )
+
+    def get_sanitized_release_channel(
+        self, release_channel: Optional[str]
+    ) -> Optional[str]:
+        """
+        Sanitize the release channel name provided by the user and validate it against the available release channels.
+
+        A return value of None indicates that release channels should not be used. Returns None if:
+        - Release channel is not provided
+        - Release channels are not enabled in the application package and the user provided the default release channel
+        """
+        if not release_channel:
+            return None
+
+        available_release_channels = get_snowflake_facade().show_release_channels(
+            self.name, self.role
+        )
+
+        if not available_release_channels and same_identifiers(
+            release_channel, DEFAULT_CHANNEL
+        ):
+            return None
+
+        self.validate_release_channel(release_channel, available_release_channels)
+        return release_channel
+
+    def validate_release_channel(
+        self,
+        release_channel: str,
+        available_release_channels: Optional[list[ReleaseChannel]] = None,
+    ) -> None:
+        """
+        Validates the release channel provided by the user and make sure it is a valid release channel for the application package.
+        """
+
+        if available_release_channels is None:
+            available_release_channels = get_snowflake_facade().show_release_channels(
+                self.name, self.role
+            )
+        if not available_release_channels:
+            raise UsageError(
+                f"Release channels are not enabled for application package {self.name}."
+            )
+        for channel in available_release_channels:
+            if same_identifiers(release_channel, channel["name"]):
+                return
+
+        raise UsageError(
+            f"Release channel {release_channel} is not available in application package {self.name}. "
+            f"Available release channels are: ({', '.join(channel['name'] for channel in available_release_channels)})."
+        )
+
     def action_release_directive_list(
         self,
         action_ctx: ActionContext,
@@ -636,25 +698,7 @@ def action_release_directive_list(
 
         If `like` is provided, only release directives matching the SQL LIKE pattern are listed.
         """
-        available_release_channels = get_snowflake_facade().show_release_channels(
-            self.name, self.role
-        )
-
-        # assume no release channel used if user selects default channel and release channels are not enabled
-        if (
-            release_channel
-            and same_identifiers(release_channel, DEFAULT_CHANNEL)
-            and not available_release_channels
-        ):
-            release_channel = None
-
-        release_channel_names = [c.get("name") for c in available_release_channels]
-        if release_channel and not identifier_in_list(
-            release_channel, release_channel_names
-        ):
-            raise ClickException(
-                f"Release channel {release_channel} does not exist in application package {self.name}."
-            )
+        release_channel = self.get_sanitized_release_channel(release_channel)
 
         release_directives = get_snowflake_facade().show_release_directives(
             package_name=self.name,
@@ -686,32 +730,15 @@ def action_release_directive_set(
         For non-default release directives, update the existing release directive if target accounts are not provided.
         """
         if target_accounts:
-            for account in target_accounts:
-                if not re.fullmatch(
-                    f"{VALID_IDENTIFIER_REGEX}\\.{VALID_IDENTIFIER_REGEX}", account
-                ):
-                    raise ClickException(
-                        f"Target account {account} is not in a valid format. Make sure you provide the target account in the format 'org.account'."
-                    )
+            self._validate_target_accounts(target_accounts)
 
         if target_accounts and same_identifiers(release_directive, DEFAULT_DIRECTIVE):
             raise BadOptionUsage(
                 "target_accounts",
                 "Target accounts can only be specified for non-default named release directives.",
             )
 
-        available_release_channels = get_snowflake_facade().show_release_channels(
-            self.name, self.role
-        )
-
-        release_channel_names = [c.get("name") for c in available_release_channels]
-
-        if not same_identifiers(
-            release_channel, DEFAULT_CHANNEL
-        ) and not identifier_in_list(release_channel, release_channel_names):
-            raise ClickException(
-                f"Release channel {release_channel} does not exist in application package {self.name}."
-            )
+        sanitized_release_channel = self.get_sanitized_release_channel(release_channel)
 
         if (
             not same_identifiers(release_directive, DEFAULT_DIRECTIVE)
@@ -722,7 +749,7 @@ def action_release_directive_set(
             get_snowflake_facade().modify_release_directive(
                 package_name=self.name,
                 release_directive=release_directive,
-                release_channel=release_channel,
+                release_channel=sanitized_release_channel,
                 version=version,
                 patch=patch,
                 role=self.role,
@@ -731,15 +758,18 @@ def action_release_directive_set(
             get_snowflake_facade().set_release_directive(
                 package_name=self.name,
                 release_directive=release_directive,
-                release_channel=release_channel if available_release_channels else None,
+                release_channel=sanitized_release_channel,
                 target_accounts=target_accounts,
                 version=version,
                 patch=patch,
                 role=self.role,
             )
 
     def action_release_directive_unset(
-        self, action_ctx: ActionContext, release_directive: str, release_channel: str
+        self,
+        action_ctx: ActionContext,
+        release_directive: str,
+        release_channel: str,
     ):
         """
         Unsets a release directive from the specified release channel.
@@ -749,21 +779,10 @@ def action_release_directive_unset(
                 "Cannot unset default release directive. Please specify a non-default release directive."
             )
 
-        available_release_channels = get_snowflake_facade().show_release_channels(
-            self.name, self.role
-        )
-        release_channel_names = [c.get("name") for c in available_release_channels]
-        if not same_identifiers(
-            release_channel, DEFAULT_CHANNEL
-        ) and not identifier_in_list(release_channel, release_channel_names):
-            raise ClickException(
-                f"Release channel {release_channel} does not exist in application package {self.name}."
-            )
-
         get_snowflake_facade().unset_release_directive(
             package_name=self.name,
             release_directive=release_directive,
-            release_channel=release_channel if available_release_channels else None,
+            release_channel=self.get_sanitized_release_channel(release_channel),
             role=self.role,
         )
 
@@ -861,6 +880,56 @@ def _bundle(self, action_ctx: ActionContext = None):
 
         return bundle_map
 
+    def action_release_channel_add_accounts(
+        self,
+        action_ctx: ActionContext,
+        release_channel: str,
+        target_accounts: list[str],
+        *args,
+        **kwargs,
+    ):
+        """
+        Adds target accounts to a release channel.
+        """
+
+        if not target_accounts:
+            raise ClickException("No target accounts provided.")
+
+        self.validate_release_channel(release_channel)
+        self._validate_target_accounts(target_accounts)
+
+        get_snowflake_facade().add_accounts_to_release_channel(
+            package_name=self.name,
+            release_channel=release_channel,
+            target_accounts=target_accounts,
+            role=self.role,
+        )
+
+    def action_release_channel_remove_accounts(
+        self,
+        action_ctx: ActionContext,
+        release_channel: str,
+        target_accounts: list[str],
+        *args,
+        **kwargs,
+    ):
+        """
+        Removes target accounts from a release channel.
+        """
+
+        if not target_accounts:
+            raise ClickException("No target accounts provided.")
+
+        self.validate_release_channel(release_channel)
+        self._validate_target_accounts(target_accounts)
+
+        get_snowflake_facade().remove_accounts_from_release_channel(
+            package_name=self.name,
+            release_channel=release_channel,
+            target_accounts=target_accounts,
+            role=self.role,
+        )
+
     def _bundle_children(self, action_ctx: ActionContext) -> List[str]:
         # Create _children directory
         children_artifacts_dir = self.children_artifacts_deploy_root