Add feature flag for enable_release_channels

src/snowflake/cli/_plugins/connection/util.py

@@ -19,7 +19,6 @@
 import os
 from enum import Enum
 from functools import lru_cache
-from textwrap import dedent
 from typing import Any, Dict, Optional
 
 from click.exceptions import ClickException
@@ -57,11 +56,12 @@ class UIParameter(Enum):
     NA_ENFORCE_MANDATORY_FILTERS = (
         "ENFORCE_MANDATORY_FILTERS_FOR_SAME_ACCOUNT_INSTALLATION"
     )
+    NA_FEATURE_RELEASE_CHANNELS = "FEATURE_RELEASE_CHANNELS"
 
 
 def get_ui_parameter(
     conn: SnowflakeConnection, parameter: UIParameter, default: Any
-) -> str:
+) -> Any:
     """
     Returns the value of a single UI parameter.
     If the parameter is not found, the default value is returned.
@@ -77,21 +77,19 @@ def get_ui_parameters(conn: SnowflakeConnection) -> Dict[UIParameter, Any]:
     Returns the UI parameters from the SYSTEM$BOOTSTRAP_DATA_REQUEST function
     """
 
-    parameters_to_fetch = sorted([param.value for param in UIParameter])
+    parameters_to_fetch = [param.value for param in UIParameter]
 
-    query = dedent(
-        f"""
-        select value['value']::string as PARAM_VALUE, value['name']::string as PARAM_NAME from table(flatten(
-            input => parse_json(SYSTEM$BOOTSTRAP_DATA_REQUEST()),
-            path => 'clientParamsInfo'
-        )) where value['name'] in ('{"', '".join(parameters_to_fetch)}');
-        """
-    )
+    # Parsing of the Json and the filtering is happening here in Snowflake CLI
+    # in order to avoid requiring a warehouse in Snowflake
+    query = "call system$bootstrap_data_request('CLIENT_PARAMS_INFO')"
+    *_, cursor = conn.execute_string(query)
 
-    *_, cursor = conn.execute_string(query, cursor_class=DictCursor)
+    json_map = json.loads(cursor.fetchone()[0])
 
     return {
-        UIParameter(row["PARAM_NAME"]): row["PARAM_VALUE"] for row in cursor.fetchall()
+        UIParameter(row["name"]): row["value"]
+        for row in json_map["clientParamsInfo"]
+        if row["name"] in parameters_to_fetch
     }
 
 
@@ -103,12 +101,7 @@ def is_regionless_redirect(conn: SnowflakeConnection) -> bool:
     assume it's regionless, as this is true for most production deployments.
     """
     try:
-        return (
-            get_ui_parameter(
-                conn, UIParameter.NA_ENABLE_REGIONLESS_REDIRECT, "true"
-            ).lower()
-            == "true"
-        )
+        return get_ui_parameter(conn, UIParameter.NA_ENABLE_REGIONLESS_REDIRECT, True)
     except:
         log.warning(
             "Cannot determine regionless redirect; assuming True.", exc_info=True

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

@@ -13,7 +13,6 @@
 from pydantic import Field, field_validator
 from snowflake.cli._plugins.connection.util import (
     UIParameter,
-    get_ui_parameter,
     make_snowsight_url,
 )
 from snowflake.cli._plugins.nativeapp.artifacts import (
@@ -124,18 +123,12 @@ def __init__(
         self._is_dev_mode = install_method.is_dev_mode
         self._metrics = get_cli_context().metrics
         self._console = console
-        connection = get_sql_executor()._conn  # noqa: SLF001
-        self._event_sharing_enabled = (
-            get_ui_parameter(
-                connection, UIParameter.NA_EVENT_SHARING_V2, "true"
-            ).lower()
-            == "true"
+
+        self._event_sharing_enabled = get_snowflake_facade().get_ui_parameter(
+            UIParameter.NA_EVENT_SHARING_V2, True
         )
-        self._event_sharing_enforced = (
-            get_ui_parameter(
-                connection, UIParameter.NA_ENFORCE_MANDATORY_FILTERS, "true"
-            ).lower()
-            == "true"
+        self._event_sharing_enforced = get_snowflake_facade().get_ui_parameter(
+            UIParameter.NA_ENFORCE_MANDATORY_FILTERS, True
         )
 
         self._share_mandatory_events = (

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

@@ -9,6 +9,7 @@
 import typer
 from click import BadOptionUsage, ClickException
 from pydantic import Field, field_validator
+from snowflake.cli._plugins.connection.util import UIParameter
 from snowflake.cli._plugins.nativeapp.artifacts import (
     BundleMap,
     VersionInfo,
@@ -25,7 +26,6 @@
     NAME_COL,
     OWNER_COL,
     PATCH_COL,
-    SPECIAL_COMMENT,
     VERSION_COL,
 )
 from snowflake.cli._plugins.nativeapp.exceptions import (
@@ -35,6 +35,7 @@
     ObjectPropertyNotFoundError,
     SetupScriptFailedValidation,
 )
+from snowflake.cli._plugins.nativeapp.feature_flags import FeatureFlag
 from snowflake.cli._plugins.nativeapp.policy import (
     AllowAlwaysPolicy,
     AskAlwaysPolicy,
@@ -824,6 +825,28 @@ def verify_project_distribution(
             return False
         return True
 
+    def _get_enable_release_channels_flag(self) -> Optional[bool]:
+        """
+        Returns the requested value of enable_release_channels flag for the application package.
+        It retrieves the value from the configuration file and checks that the feature is enabled in the account.
+        If return value is None, it means do not explicitly set the flag.
+        """
+        feature_flag_from_config = FeatureFlag.ENABLE_RELEASE_CHANNELS.get_value()
+        feature_enabled_in_account = (
+            get_snowflake_facade().get_ui_parameter(
+                UIParameter.NA_FEATURE_RELEASE_CHANNELS, "ENABLED"
+            )
+            == "ENABLED"
+        )
+
+        if feature_flag_from_config is not None and not feature_enabled_in_account:
+            self._workspace_ctx.console.warning(
+                f"Ignoring feature flag {FeatureFlag.ENABLE_RELEASE_CHANNELS.name} because release channels are not enabled in the current account."
+            )
+            return None
+
+        return feature_flag_from_config
+
     def create_app_package(self) -> None:
         """
         Creates the application package with our up-to-date stage if none exists.
@@ -851,21 +874,23 @@ def create_app_package(self) -> None:
                 if row_comment not in ALLOWED_SPECIAL_COMMENTS:
                     raise ApplicationPackageAlreadyExistsError(self.name)
 
+            # 4. Update the application package with setting enable_release_channels if necessary
+            get_snowflake_facade().alter_application_package_properties(
+                package_name=self.name,
+                enable_release_channels=self._get_enable_release_channels_flag(),
+                role=self.role,
+            )
+
             return
 
         # If no application package pre-exists, create an application package, with the specified distribution in the project definition file.
-        sql_executor = get_sql_executor()
-        with sql_executor.use_role(self.role):
-            console.step(f"Creating new application package {self.name} in account.")
-            sql_executor.execute_query(
-                dedent(
-                    f"""\
-                    create application package {self.name}
-                        comment = {SPECIAL_COMMENT}
-                        distribution = {model.distribution}
-                """
-                )
-            )
+        console.step(f"Creating new application package {self.name} in account.")
+        get_snowflake_facade().create_application_package(
+            role=self.role,
+            enable_release_channels=self._get_enable_release_channels_flag(),
+            distribution=model.distribution,
+            package_name=self.name,
+        )
 
     def execute_post_deploy_hooks(self):
         execute_post_deploy_hooks(

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

@@ -22,3 +22,4 @@ class FeatureFlag(FeatureFlagMixin):
     ENABLE_NATIVE_APP_PYTHON_SETUP = BooleanFlag(
         "ENABLE_NATIVE_APP_PYTHON_SETUP", False
     )
+    ENABLE_RELEASE_CHANNELS = BooleanFlag("ENABLE_RELEASE_CHANNELS", None)

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

@@ -18,6 +18,7 @@
 from textwrap import dedent
 from typing import Any, Dict, List
 
+from snowflake.cli._plugins.connection.util import UIParameter, get_ui_parameter
 from snowflake.cli._plugins.nativeapp.constants import (
     AUTHORIZE_TELEMETRY_COL,
     NAME_COL,
@@ -49,6 +50,7 @@
 )
 from snowflake.cli.api.identifiers import FQN
 from snowflake.cli.api.metrics import CLICounterField
+from snowflake.cli.api.project.schemas.v1.native_app.package import DistributionOptions
 from snowflake.cli.api.project.util import (
     identifier_to_show_like_pattern,
     is_valid_unquoted_identifier,
@@ -743,6 +745,100 @@ def create_application(
 
             return create_cursor.fetchall()
 
+    def create_application_package(
+        self,
+        package_name: str,
+        distribution: DistributionOptions,
+        enable_release_channels: bool | None = None,
+        role: str | None = None,
+    ) -> None:
+        """
+        Creates a new application package.
+        @param package_name: Name of the application package to create.
+        @param [Optional] enable_release_channels: Enable/Disable release channels if not None.
+        @param [Optional] role: Role to switch to while running this script. Current role will be used if no role is passed in.
+        """
+        package_name = to_identifier(package_name)
+
+        enable_release_channels_clause = ""
+        if enable_release_channels is not None:
+            enable_release_channels_clause = (
+                f"enable_release_channels = {str(enable_release_channels).lower()}"
+            )
+
+        with self._use_role_optional(role):
+            try:
+                self._sql_executor.execute_query(
+                    dedent(
+                        _strip_empty_lines(
+                            f"""\
+                            create application package {package_name}
+                                comment = {SPECIAL_COMMENT}
+                                distribution = {distribution}
+                                {enable_release_channels_clause}
+                            """
+                        )
+                    )
+                )
+            except ProgrammingError as err:
+                if err.errno == INSUFFICIENT_PRIVILEGES:
+                    raise InsufficientPrivilegesError(
+                        f"Insufficient privileges to create application package {package_name}",
+                        role=role,
+                    ) from err
+                handle_unclassified_error(
+                    err, f"Failed to create application package {package_name}."
+                )
+
+    def alter_application_package_properties(
+        self,
+        package_name: str,
+        enable_release_channels: bool | None = None,
+        role: str | None = None,
+    ) -> None:
+        """
+        Alters the properties of an existing application package.
+        @param package_name: Name of the application package to alter.
+        @param [Optional] enable_release_channels: Enable/Disable release channels if not None.
+        @param [Optional] role: Role to switch to while running this script. Current role will be used if no role is passed in.
+        """
+
+        package_name = to_identifier(package_name)
+
+        if enable_release_channels is not None:
+            with self._use_role_optional(role):
+                try:
+                    self._sql_executor.execute_query(
+                        dedent(
+                            f"""\
+                            alter application package {package_name}
+                                set enable_release_channels = {str(enable_release_channels).lower()}
+                        """
+                        )
+                    )
+                except ProgrammingError as err:
+                    if err.errno == INSUFFICIENT_PRIVILEGES:
+                        raise InsufficientPrivilegesError(
+                            f"Insufficient privileges update enable_release_channels for application package {package_name}",
+                            role=role,
+                        ) from err
+                    handle_unclassified_error(
+                        err,
+                        f"Failed to update enable_release_channels for application package {package_name}.",
+                    )
+
+    def get_ui_parameter(self, parameter: UIParameter, default: Any) -> Any:
+        """
+        Returns the value of a single UI parameter.
+        If the parameter is not found, the default value is returned.
+
+        @param parameter: UIParameter, the parameter to get the value of.
+        @param default: Default value to return if the parameter is not found.
+        """
+        connection = self._sql_executor._conn  # noqa SLF001
+
+        return get_ui_parameter(connection, parameter, default)
+
 
 # TODO move this to src/snowflake/cli/api/project/util.py in a separate
 # PR since it's codeowned by the CLI team
@@ -763,3 +859,10 @@ def _same_identifier(id1: str, id2: str) -> bool:
     # The canonical identifiers are equal if they are equal when both are quoted
     # (if they are already quoted, this is a no-op)
     return to_quoted_identifier(canonical_id1) == to_quoted_identifier(canonical_id2)
+
+
+def _strip_empty_lines(text: str) -> str:
+    """
+    Strips empty lines from the input string.
+    """
+    return "\n".join(line for line in text.splitlines() if line.strip())

src/snowflake/cli/api/config.py

@@ -286,8 +286,12 @@ def get_config_value(*path, key: str, default: Optional[Any] = Empty) -> Any:
         raise
 
 
-def get_config_bool_value(*path, key: str, default: Optional[Any] = Empty) -> bool:
-    value = get_config_value(*path, key=key, default=default)
+def get_config_bool_value(*path, key: str, default: Optional[bool]) -> Optional[bool]:
+    value = get_config_value(*path, key=key, default=None)
+
+    if value is None:
+        return default
+
     try:
         return try_cast_to_bool(value)
     except ValueError:

src/snowflake/cli/api/feature_flags.py

@@ -24,20 +24,31 @@
 
 class BooleanFlag(NamedTuple):
     name: str
-    default: bool = False
+    default: bool | None = False
 
 
 @unique
 class FeatureFlagMixin(Enum):
-    def is_enabled(self) -> bool:
+    def get_value(self) -> bool | None:
         return get_config_bool_value(
             *FEATURE_FLAGS_SECTION_PATH,
             key=self.value.name.lower(),
             default=self.value.default,
         )
 
-    def is_disabled(self):
-        return not self.is_enabled()
+    def is_enabled(self) -> bool:
+        return self.get_value() is True
+
+    def is_disabled(self) -> bool:
+        return self.get_value() is False
+
+    def is_set(self) -> bool:
+        return (
+            get_config_bool_value(
+                *FEATURE_FLAGS_SECTION_PATH, key=self.value.name.lower(), default=None
+            )
+            is not None
+        )
 
     def env_variable(self):
         return get_env_variable_name(*FEATURE_FLAGS_SECTION_PATH, key=self.value.name)