adding az containerapp function keys commands

Author: khushishah513

src/containerapp/HISTORY.rst

@@ -4,6 +4,10 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp function list-keys': List function keys for a specific function in a container app
+* 'az containerapp function update-keys': Update specific function key for a specific function in a container app
+* 'az containerapp function list-hostkeys': List host keys for a container app
+* 'az containerapp function update-hostkeys': Update specific host key for a container app
 * 'az containerapp update/up': Disallow changing `--revisions-mode` to Labels.
 * 'az containerapp session code-interpreter': Fix `--path` in examples
 * 'az containerapp sessionpool create/update': Support `--lifecycle-type` and `--max-alive-period`

src/containerapp/azext_containerapp/_clients.py

@@ -1698,3 +1698,79 @@ def remove(cls, cmd, resource_group_name, environment_name):
             if r.status_code == 202:
                 operation_url = r.headers.get(HEADER_LOCATION)
                 poll_results(cmd, operation_url)
+
+    class ContainerAppFunctionsPreviewClient:
+    api_version = PREVIEW_API_VERSION
+
+    @classmethod
+    def list_function_keys(cls, cmd, resource_group_name, name, function_name):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/containerapps/{}/functions/{}/listkeys?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            function_name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "POST", request_url)
+        return r.json()
+
+    @classmethod
+    def update_function_keys(cls, cmd, resource_group_name, name, function_name, key_name, key_value=None):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/containerapps/{}/functions/{}/keys/{}?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            function_name,
+            key_name,
+            cls.api_version)
+
+        body = {}
+        if key_value:
+            body["value"] = key_value
+
+        r = send_raw_request(cmd.cli_ctx, "PUT", request_url, body=json.dumps(body))
+        return r.json()
+
+    @classmethod
+    def list_host_keys(cls, cmd, resource_group_name, name):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/containerapps/{}/host/default/listkeys?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "POST", request_url)
+        return r.json()
+
+    @classmethod
+    def update_host_keys(cls, cmd, resource_group_name, name, key_type, key_name, key_value=None):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/containerapps/{}/host/default/{}/{}?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            key_type,
+            key_name,
+            cls.api_version)
+
+        body = {}
+        if key_value:
+            body["value"] = key_value
+
+        r = send_raw_request(cmd.cli_ctx, "PUT", request_url, body=json.dumps(body))
+        return r.json()

src/containerapp/azext_containerapp/_help.py

@@ -166,7 +166,6 @@
             az containerapp up -n my-containerapp --image my-app:v1.0 --kind functionapp
 """
 
-
 helps['containerapp replica count'] = """
     type: command
     short-summary: Count of a container app's replica(s)
@@ -179,6 +178,53 @@
           az containerapp replica count -n my-containerapp -g MyResourceGroup
 """
 
+helps['containerapp function'] = """
+    type: group
+    short-summary: Commands to manage function keys in a container app.
+"""
+
+helps['containerapp function list-keys'] = """
+    type: command
+    short-summary: List function keys for a specific function in a container app.
+    examples:
+    - name: List function keys for a specific function
+      text: |
+          az containerapp function list-keys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision --function-name MyFunctionName
+"""
+
+helps['containerapp function update-keys'] = """
+    type: command
+    short-summary: Update function keys for a specific function in a container app.
+    examples:
+    - name: Update a function key for a specific function
+      text: |
+          az containerapp function update-keys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision --function-name MyFunctionName --key-name MyKeyName --key-value MyKeyValue
+"""
+
+helps['containerapp function list-hostkeys'] = """
+    type: command
+    short-summary: List host keys for a container app.
+    examples:
+    - name: List host keys for a container app
+      text: |
+          az containerapp function list-hostkeys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision
+"""
+
+helps['containerapp function update-hostkeys'] = """
+    type: command
+    short-summary: Update host keys for a container app.
+    examples:
+    - name: Update a host key for a container app with function key type
+      text: |
+          az containerapp function update-hostkeys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision --key-name MyKeyName --key-value MyKeyValue --key-type functionKeys
+    - name: Update a host key for a container app with master key type
+      text: |
+          az containerapp function update-hostkeys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision --key-name MyKeyName --key-value MyKeyValue --key-type masterKey
+    - name: Update a host key for a container app with system key type
+      text: |
+          az containerapp function update-hostkeys -n my-containerapp -g MyResourceGroup --revision MyContainerappRevision --key-name MyKeyName --key-value MyKeyValue --key-type systemKeys
+"""
+
 # Environment Commands
 helps['containerapp env'] = """
     type: group

src/containerapp/azext_containerapp/_params.py

@@ -529,3 +529,30 @@ def load_arguments(self, _):
         c.argument('termination_grace_period', options_list=['--termination-grace-period', '-t'], type=int, help="Time in seconds to drain requests during ingress shutdown. Default 500, minimum 0, maximum 3600.")
         c.argument('request_idle_timeout', options_list=['--request-idle-timeout'], type=int, help="Timeout in minutes for idle requests. Default 4, minimum 4, maximum 30.")
         c.argument('header_count_limit', options_list=['--header-count-limit'], type=int, help="Limit of http headers per request. Default 100, minimum 1.")
+
+    with self.argument_context('containerapp function list-keys') as c:
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+        c.argument('name', options_list=['--name', '-n'], help="The name of the container app.")
+        c.argument('revision', options_list=['--revision'], help="The name of the container app revision.")
+        c.argument('function_name', options_list=['--function-name'], help="The name of the function.")
+
+    with self.argument_context('containerapp function update-keys') as c:
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+        c.argument('name', options_list=['--name', '-n'], help="The name of the container app.")
+        c.argument('revision', options_list=['--revision'], help="The name of the container app revision.")
+        c.argument('function_name', options_list=['--function-name'], help="The name of the function.")
+        c.argument('key_name', options_list=['--key-name'], help="The name of the key to update.")
+        c.argument('key_value', options_list=['--key-value'], help="The value of the key to update.")
+
+    with self.argument_context('containerapp function list-hostkeys') as c:
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+        c.argument('name', options_list=['--name', '-n'], help="The name of the container app.")
+        c.argument('revision', options_list=['--revision'], help="The name of the container app revision.")
+
+    with self.argument_context('containerapp function update-hostkeys') as c:
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+        c.argument('name', options_list=['--name', '-n'], help="The name of the container app.")
+        c.argument('revision', options_list=['--revision'], help="The name of the container app revision.")
+        c.argument('key_name', options_list=['--key-name'], help="The name of the host key to update.")
+        c.argument('key_value', options_list=['--key-value'], help="The value of the host key to update.")
+        c.argument('key_type', options_list=['--key-type'], arg_type=get_enum_type(['functionKeys', 'masterKey', 'systemKeys']), help="The type of the host key.")

src/containerapp/azext_containerapp/commands.py

@@ -297,3 +297,9 @@ def load_command_table(self, args):
         g.custom_command('add', 'add_environment_premium_ingress')
         g.custom_command('update', 'update_environment_premium_ingress')
         g.custom_command('remove', 'remove_environment_premium_ingress', confirmation=True)
+
+    with self.command_group('containerapp function') as g:
+        g.custom_command('list-keys', 'list_containerapp_function_keys')
+        g.custom_command('update-keys', 'update_containerapp_function_keys')
+        g.custom_command('list-hostkeys', 'list_containerapp_function_hostkeys')
+        g.custom_command('update-hostkeys', 'update_containerapp_function_hostkeys')

src/containerapp/azext_containerapp/containerapp_function_keys_decorator.py

@@ -0,0 +1,179 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+# pylint: disable=line-too-long, broad-except, logging-format-interpolation
+from knack.log import get_logger
+from typing import Any, Dict
+
+from azure.cli.core.commands import AzCliCommand
+from azure.cli.core.azclierror import ValidationError
+from azure.cli.command_modules.containerapp.base_resource import BaseResource
+
+from ._clients import ContainerAppFunctionsPreviewClient
+from ._client_factory import handle_raw_exception
+
+logger = get_logger(__name__)
+
+
+class ContainerAppFunctionsDecorator(BaseResource):
+    """Base decorator for Container App Functions operations"""
+
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def get_argument_function_name(self):
+        return self.get_param("function_name")
+
+
+class ContainerAppFunctionKeysListDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for listing function keys"""
+
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def list_keys(self):
+        """List keys for a specific function"""
+        try:
+            function_name = self.get_argument_function_name()
+            resource_group_name = self.get_argument_resource_group_name()
+            name = self.get_argument_name()
+
+            if not resource_group_name:
+                raise ValidationError("Resource group name is required.")
+
+            if not name:
+                raise ValidationError("Container app name is required.")
+
+            if not function_name:
+                raise ValidationError("Function name is required.")
+
+            return self.client.list_function_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                function_name=function_name
+            )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionKeysUpdateDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for updating function keys"""
+
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def get_argument_key_name(self):
+        return self.get_param("key_name")
+
+    def get_argument_key_value(self):
+        return self.get_param("key_value")
+
+    def update_keys(self):
+        """Update keys for a specific function"""
+        try:
+            function_name = self.get_argument_function_name()
+            key_name = self.get_argument_key_name()
+            key_value = self.get_argument_key_value()
+            resource_group_name = self.get_argument_resource_group_name()
+            name = self.get_argument_name()
+
+            if not resource_group_name:
+                raise ValidationError("Resource group name is required.")
+
+            if not name:
+                raise ValidationError("Container app name is required.")
+
+            if not function_name:
+                raise ValidationError("Function name is required.")
+
+            if not key_name:
+                raise ValidationError("Key name is required.")
+
+            return self.client.update_function_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                function_name=function_name,
+                key_name=key_name,
+                key_value=key_value
+            )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionHostKeysListDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for listing host keys"""
+
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def list_host_keys(self):
+        """List host keys for the container app function host"""
+        try:
+            resource_group_name = self.get_argument_resource_group_name()
+            name = self.get_argument_name()
+
+            if not resource_group_name:
+                raise ValidationError("Resource group name is required.")
+
+            if not name:
+                raise ValidationError("Container app name is required.")
+
+            return self.client.list_host_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name
+            )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionHostKeysUpdateDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for updating host keys"""
+
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def get_argument_key_type(self):
+        return self.get_param("key_type")
+
+    def get_argument_key_name(self):
+        return self.get_param("key_name")
+
+    def get_argument_key_value(self):
+        return self.get_param("key_value")
+
+    def update_host_keys(self):
+        """Update host keys for the container app function host"""
+        try:
+            key_type = self.get_argument_key_type()
+            key_name = self.get_argument_key_name()
+            key_value = self.get_argument_key_value()
+            resource_group_name = self.get_argument_resource_group_name()
+            name = self.get_argument_name()
+
+            if not resource_group_name:
+                raise ValidationError("Resource group name is required.")
+
+            if not name:
+                raise ValidationError("Container app name is required.")
+
+            if not key_type:
+                raise ValidationError("Key type is required.")
+
+            if not key_name:
+                raise ValidationError("Key name is required.")
+
+            return self.client.update_host_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                key_type=key_type,
+                key_name=key_name,
+                key_value=key_value
+            )
+        except Exception as e:
+            handle_raw_exception(e)