Add release directives support to SnowCLI

RELEASE-NOTES.md

@@ -19,6 +19,7 @@
 ## Deprecations
 
 ## New additions
+* Add Release Directives support, through `snow app release-directive` commands.
 
 ## Fixes and improvements
 * Fixed crashes with older x86_64 Intel CPUs.

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

@@ -31,6 +31,9 @@
 from snowflake.cli._plugins.nativeapp.entities.application_package import (
     ApplicationPackageEntityModel,
 )
+from snowflake.cli._plugins.nativeapp.release_directive.commands import (
+    app as release_directives_app,
+)
 from snowflake.cli._plugins.nativeapp.sf_facade import get_snowflake_facade
 from snowflake.cli._plugins.nativeapp.v2_conversions.compat import (
     find_entity,
@@ -67,6 +70,7 @@
     help="Manages a Snowflake Native App",
 )
 app.add_typer(versions_app)
+app.add_typer(release_directives_app)
 
 log = logging.getLogger(__name__)
 

src/snowflake/cli/_plugins/nativeapp/constants.py

@@ -22,6 +22,7 @@
 OWNER_COL = "owner"
 VERSION_COL = "version"
 PATCH_COL = "patch"
+CHANNEL_COL = "release_channel_name"
 AUTHORIZE_TELEMETRY_COL = "authorize_telemetry_event_sharing"
 
 INTERNAL_DISTRIBUTION = "internal"

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

@@ -4,7 +4,7 @@
 import re
 from pathlib import Path
 from textwrap import dedent
-from typing import List, Literal, Optional, Union
+from typing import Any, List, Literal, Optional, Union
 
 import typer
 from click import BadOptionUsage, ClickException
@@ -78,9 +78,13 @@
 from snowflake.cli.api.project.schemas.v1.native_app.path_mapping import PathMapping
 from snowflake.cli.api.project.util import (
     SCHEMA_AND_NAME,
+    VALID_IDENTIFIER_REGEX,
     append_test_resource_suffix,
     extract_schema,
+    identifier_in_list,
     identifier_to_show_like_pattern,
+    same_identifiers,
+    sql_match,
     to_identifier,
     unquote_identifier,
 )
@@ -544,6 +548,148 @@ def action_version_drop(
             f"Version {version} in application package {self.name} dropped successfully."
         )
 
+    def action_release_directive_list(
+        self,
+        action_ctx: ActionContext,
+        release_channel: Optional[str],
+        like: str,
+        *args,
+        **kwargs,
+    ) -> list[dict[str, Any]]:
+        """
+        Get all existing release directives for an application package.
+        Limit the results to a specific release channel, if provided.
+
+        If `like` is provided, only release directives matching the SQL LIKE pattern are listed.
+        """
+        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")
+            and not release_channels
+        ):
+            release_channel = None
+
+        release_channel_names = [c.get("name") for c in 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_directives = get_snowflake_facade().show_release_directives(
+            package_name=self.name,
+            role=self.role,
+            release_channel=release_channel,
+        )
+
+        return [
+            directive
+            for directive in release_directives
+            if sql_match(pattern=like, value=directive.get("name", ""))
+        ]
+
+    def action_release_directive_set(
+        self,
+        action_ctx: ActionContext,
+        version: str,
+        patch: int,
+        release_directive: str,
+        release_channel: str,
+        target_accounts: Optional[list[str]],
+        *args,
+        **kwargs,
+    ):
+        """
+        Sets a release directive to the specified version and patch using the specified release channel.
+        Target accounts can only be specified for non-default release directives.
+
+        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'."
+                    )
+
+        if target_accounts and same_identifiers(release_directive, "DEFAULT"):
+            raise BadOptionUsage(
+                "target_accounts",
+                "Target accounts can only be specified for non-default named release directives.",
+            )
+
+        release_channels = get_snowflake_facade().show_release_channels(
+            self.name, self.role
+        )
+
+        release_channel_names = [c.get("name") for c in release_channels]
+
+        if not same_identifiers(release_channel, "DEFAULT") 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}."
+            )
+
+        if not same_identifiers(release_directive, "DEFAULT") and not target_accounts:
+            # if it is a non-default release directive with no target accounts specified,
+            # it means that the user wants to modify existing release directive
+            get_snowflake_facade().modify_release_directive(
+                package_name=self.name,
+                release_directive=release_directive,
+                release_channel=release_channel,
+                version=version,
+                patch=patch,
+                role=self.role,
+            )
+        else:
+            get_snowflake_facade().set_release_directive(
+                package_name=self.name,
+                release_directive=release_directive,
+                release_channel=release_channel if release_channels else None,
+                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
+    ):
+        """
+        Unsets a release directive from the specified release channel.
+        """
+        if same_identifiers(release_directive, "DEFAULT"):
+            raise ClickException(
+                "Cannot unset default release directive. Please specify a non-default release directive."
+            )
+
+        release_channels = get_snowflake_facade().show_release_channels(
+            self.name, self.role
+        )
+        release_channel_names = [c.get("name") for c in release_channels]
+        if not same_identifiers(release_channel, "DEFAULT") 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 release_channels else None,
+            role=self.role,
+        )
+
     def _bundle(self):
         model = self._entity_model
         bundle_map = build_bundle(self.project_root, self.deploy_root, model.artifacts)
@@ -664,7 +810,7 @@ def get_existing_release_directive_info_for_version(
         It executes a 'show release directives in application package' query and returns the filtered results, if they exist.
         """
         release_directives = get_snowflake_facade().show_release_directives(
-            self.name, self.role
+            package_name=self.name, role=self.role
         )
         return [
             directive

src/snowflake/cli/_plugins/nativeapp/release_directive/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) 2024 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

src/snowflake/cli/_plugins/nativeapp/release_directive/commands.py

@@ -0,0 +1,164 @@
+# Copyright (c) 2024 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+import typer
+from snowflake.cli._plugins.nativeapp.v2_conversions.compat import (
+    force_project_definition_v2,
+)
+from snowflake.cli._plugins.workspace.manager import WorkspaceManager
+from snowflake.cli.api.cli_global_context import get_cli_context
+from snowflake.cli.api.commands.decorators import with_project_definition
+from snowflake.cli.api.commands.flags import like_option
+from snowflake.cli.api.commands.snow_typer import SnowTyperFactory
+from snowflake.cli.api.entities.common import EntityActions
+from snowflake.cli.api.output.types import (
+    CollectionResult,
+    CommandResult,
+    MessageResult,
+)
+
+app = SnowTyperFactory(
+    name="release-directive",
+    help="Manages release directives of an application package",
+)
+
+log = logging.getLogger(__name__)
+
+
+@app.command("list", requires_connection=True)
+@with_project_definition()
+@force_project_definition_v2()
+def release_directive_list(
+    like: str = like_option(
+        help_example="`snow app release-directive list --like='my%'` lists all release directives starting with 'my'",
+    ),
+    channel: Optional[str] = typer.Option(
+        default=None,
+        show_default=False,
+        help="The release channel to use when listing release directives. If not provided, release directives from all release channels are listed.",
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Lists release directives in an application package.
+    If no release channel is specified, release directives for all channels are listed.
+    If a release channel is specified, only release directives for that channel are listed.
+
+    If `--like` is provided, only release directives matching the SQL pattern are listed.
+    """
+
+    cli_context = get_cli_context()
+    ws = WorkspaceManager(
+        project_definition=cli_context.project_definition,
+        project_root=cli_context.project_root,
+    )
+    package_id = options["package_entity_id"]
+    result = ws.perform_action(
+        package_id,
+        EntityActions.RELEASE_DIRECTIVE_LIST,
+        release_channel=channel,
+        like=like,
+    )
+
+    return CollectionResult(result)
+
+
+@app.command("set", requires_connection=True)
+@with_project_definition()
+@force_project_definition_v2()
+def release_directive_set(
+    directive: str = typer.Argument(
+        show_default=False,
+        help="Name of the release directive to set",
+    ),
+    channel: str = typer.Option(
+        "DEFAULT",
+        help="Name of the release channel to use",
+    ),
+    target_accounts: Optional[list[str]] = typer.Option(
+        None,
+        show_default=False,
+        help="List of the accounts to apply the release directive to. Format has to be `org1.account1,org2.account2`",
+    ),
+    version: str = typer.Option(
+        show_default=False,
+        help="Version of the application package to use",
+    ),
+    patch: int = typer.Option(
+        show_default=False,
+        help="Patch number to use for the selected version",
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Sets a release directive.
+
+    target_accounts cannot be specified for default release directives.
+    target_accounts field is required when creating a new non-default release directive.
+    """
+
+    cli_context = get_cli_context()
+    ws = WorkspaceManager(
+        project_definition=cli_context.project_definition,
+        project_root=cli_context.project_root,
+    )
+    package_id = options["package_entity_id"]
+    ws.perform_action(
+        package_id,
+        EntityActions.RELEASE_DIRECTIVE_SET,
+        release_directive=directive,
+        version=version,
+        patch=patch,
+        target_accounts=target_accounts,
+        release_channel=channel,
+    )
+    return MessageResult("Successfully set release directive.")
+
+
+@app.command("unset", requires_connection=True)
+@with_project_definition()
+@force_project_definition_v2()
+def release_directive_unset(
+    directive: str = typer.Argument(
+        show_default=False,
+        help="Name of the release directive",
+    ),
+    channel: Optional[str] = typer.Option(
+        "DEFAULT",
+        help="Name of the release channel to use",
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Unsets a release directive.
+    """
+
+    cli_context = get_cli_context()
+    ws = WorkspaceManager(
+        project_definition=cli_context.project_definition,
+        project_root=cli_context.project_root,
+    )
+    package_id = options["package_entity_id"]
+    ws.perform_action(
+        package_id,
+        EntityActions.RELEASE_DIRECTIVE_UNSET,
+        release_directive=directive,
+        release_channel=channel,
+    )
+    return MessageResult(f"Successfully unset release directive {directive}.")