Add release channels list support

RELEASE-NOTES.md

@@ -26,6 +26,7 @@
 * Add support for release channels feature in native app version creation/drop.
 * `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
 
 ## 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_channel.commands import (
+    app as release_channels_app,
+)
 from snowflake.cli._plugins.nativeapp.release_directive.commands import (
     app as release_directives_app,
 )
@@ -71,6 +74,7 @@
 )
 app.add_typer(versions_app)
 app.add_typer(release_directives_app)
+app.add_typer(release_channels_app)
 
 log = logging.getLogger(__name__)
 

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

@@ -53,6 +53,7 @@
 from snowflake.cli._plugins.nativeapp.sf_facade_exceptions import (
     InsufficientPrivilegesError,
 )
+from snowflake.cli._plugins.nativeapp.sf_sql_facade import ReleaseChannel
 from snowflake.cli._plugins.nativeapp.utils import needs_confirmation, sanitize_dir_name
 from snowflake.cli._plugins.snowpark.snowpark_entity_model import (
     FunctionEntityModel,
@@ -766,6 +767,69 @@ def action_release_directive_unset(
             role=self.role,
         )
 
+    def _print_channel_to_console(self, channel: ReleaseChannel) -> None:
+        """
+        Prints the release channel details to the console.
+        """
+        console = self._workspace_ctx.console
+
+        console.message(f"""[bold]{channel["name"]}[/bold]""")
+        accounts_list: Optional[list[str]] = channel["targets"].get("accounts")
+        target_accounts = (
+            f"({', '.join(accounts_list)})"
+            if accounts_list is not None
+            else "ALL ACCOUNTS"
+        )
+
+        formatted_created_on = (
+            channel["created_on"].astimezone().strftime("%Y-%m-%d %H:%M:%S.%f %Z")
+            if channel["created_on"]
+            else ""
+        )
+
+        formatted_updated_on = (
+            channel["updated_on"].astimezone().strftime("%Y-%m-%d %H:%M:%S.%f %Z")
+            if channel["updated_on"]
+            else ""
+        )
+        with console.indented():
+            console.message(f"Description: {channel['description']}")
+            console.message(f"Versions: ({', '.join(channel['versions'])})")
+            console.message(f"Created on: {formatted_created_on}")
+            console.message(f"Updated on: {formatted_updated_on}")
+            console.message(f"Target accounts: {target_accounts}")
+
+    def action_release_channel_list(
+        self,
+        action_ctx: ActionContext,
+        release_channel: Optional[str],
+        *args,
+        **kwargs,
+    ) -> list[ReleaseChannel]:
+        """
+        Get all existing release channels for an application package.
+        If `release_channel` is provided, only the specified release channel is listed.
+        """
+        console = self._workspace_ctx.console
+        available_channels = get_snowflake_facade().show_release_channels(
+            self.name, self.role
+        )
+
+        filtered_channels = [
+            channel
+            for channel in available_channels
+            if release_channel is None
+            or same_identifiers(channel["name"], release_channel)
+        ]
+
+        if not filtered_channels:
+            console.message("No release channels found.")
+        else:
+            for channel in filtered_channels:
+                self._print_channel_to_console(channel)
+
+        return filtered_channels
+
     def _bundle(self, action_ctx: ActionContext = None):
         model = self._entity_model
         bundle_map = build_bundle(self.project_root, self.deploy_root, model.artifacts)

src/snowflake/cli/_plugins/nativeapp/release_channel/__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_channel/commands.py

@@ -0,0 +1,71 @@
+# 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.snow_typer import SnowTyperFactory
+from snowflake.cli.api.entities.common import EntityActions
+from snowflake.cli.api.output.formats import OutputFormat
+from snowflake.cli.api.output.types import (
+    CollectionResult,
+    CommandResult,
+)
+
+app = SnowTyperFactory(
+    name="release-channel",
+    help="Manages release channels of an application package",
+)
+
+log = logging.getLogger(__name__)
+
+
+@app.command("list", requires_connection=True)
+@with_project_definition()
+@force_project_definition_v2()
+def release_channel_list(
+    channel: Optional[str] = typer.Argument(
+        default=None,
+        show_default=False,
+        help="The release channel to list. If not provided, all release channels are listed.",
+    ),
+    **options,
+) -> CommandResult:
+    """
+    Lists the release channels available for an application package.
+    """
+
+    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"]
+    channels = ws.perform_action(
+        package_id,
+        EntityActions.RELEASE_CHANNEL_LIST,
+        release_channel=channel,
+    )
+
+    if cli_context.output_format == OutputFormat.JSON:
+        return CollectionResult(channels)

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

@@ -13,11 +13,13 @@
 # limitations under the License.
 from __future__ import annotations
 
+import json
 import logging
 from contextlib import contextmanager
+from datetime import datetime
 from functools import cache
 from textwrap import dedent
-from typing import Any, Dict, List
+from typing import Any, Dict, List, TypedDict
 
 from snowflake.cli._plugins.connection.util import UIParameter, get_ui_parameter
 from snowflake.cli._plugins.nativeapp.constants import (
@@ -52,6 +54,7 @@
     CANNOT_DISABLE_MANDATORY_TELEMETRY,
     CANNOT_DISABLE_RELEASE_CHANNELS,
     DOES_NOT_EXIST_OR_CANNOT_BE_PERFORMED,
+    DOES_NOT_EXIST_OR_NOT_AUTHORIZED,
     INSUFFICIENT_PRIVILEGES,
     NO_WAREHOUSE_SELECTED_IN_SESSION,
     RELEASE_DIRECTIVE_DOES_NOT_EXIST,
@@ -73,6 +76,18 @@
 from snowflake.cli.api.utils.cursor import find_first_row
 from snowflake.connector import DictCursor, ProgrammingError
 
+ReleaseChannel = TypedDict(
+    "ReleaseChannel",
+    {
+        "name": str,
+        "description": str,
+        "created_on": datetime,
+        "updated_on": datetime,
+        "targets": dict[str, Any],
+        "versions": list[str],
+    },
+)
+
 
 class SnowflakeSQLFacade:
     def __init__(self, sql_executor: BaseSqlExecutor | None = None):
@@ -1141,7 +1156,7 @@ def unset_release_directive(
 
     def show_release_channels(
         self, package_name: str, role: str | None = None
-    ) -> list[dict[str, Any]]:
+    ) -> list[ReleaseChannel]:
         """
         Show release channels in a package.
         @param package_name: Name of the package
@@ -1155,6 +1170,7 @@ def show_release_channels(
             return []
 
         package_identifier = to_identifier(package_name)
+        results = []
         with self._use_role_optional(role):
             try:
                 cursor = self._sql_executor.execute_query(
@@ -1166,11 +1182,31 @@ def show_release_channels(
                 if err.errno == SQL_COMPILATION_ERROR:
                     # Release not out yet and param not out yet
                     return []
+                if err.errno == DOES_NOT_EXIST_OR_NOT_AUTHORIZED:
+                    raise UserInputError(
+                        f"Application package {package_name} does not exist or you are not authorized to access it."
+                    ) from err
                 handle_unclassified_error(
                     err,
                     f"Failed to show release channels for application package {package_name}.",
                 )
-            return cursor.fetchall()
+            rows = cursor.fetchall()
+
+            for row in rows:
+                targets = json.loads(row["targets"]) if row.get("targets") else {}
+                versions = json.loads(row["versions"]) if row.get("versions") else []
+                results.append(
+                    ReleaseChannel(
+                        name=row["name"],
+                        description=row["description"],
+                        created_on=row["created_on"],
+                        updated_on=row["updated_on"],
+                        targets=targets,
+                        versions=versions,
+                    )
+                )
+
+            return results
 
 
 def _strip_empty_lines(text: str) -> str: