adding az containerapp functions commands

Author: khushishah513

src/containerapp/HISTORY.rst

@@ -4,6 +4,7 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp functions get/list': New command group to list and show functions in container apps.
 * '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

@@ -303,6 +303,78 @@ def list(cls, cmd, resource_group_name, container_app_name):
         return policy_list
 
 
+class ContainerAppFunctionsPreviewClient():
+    api_version = PREVIEW_API_VERSION
+
+    @classmethod
+    def list_functions_by_revision(cls, cmd, resource_group_name, container_app_name, revision_name):
+        """List all functions for a specific revision"""
+        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/{}/revisions/{}/functions?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            container_app_name,
+            revision_name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "GET", request_url)
+        return r.json()
+
+    @classmethod
+    def get_function_by_revision(cls, cmd, resource_group_name, container_app_name, revision_name, function_name):
+        """Get a specific function for a specific revision"""
+        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/{}/revisions/{}/functions/{}?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            container_app_name,
+            revision_name,
+            function_name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "GET", request_url)
+        return r.json()
+
+    @classmethod
+    def list_functions(cls, cmd, resource_group_name, container_app_name):
+        """List all functions for a container app (across all revisions)"""
+        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?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            container_app_name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "GET", request_url)
+        return r.json()
+
+    @classmethod
+    def get_function(cls, cmd, resource_group_name, container_app_name, function_name):
+        """Get a specific function for a container app"""
+        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/{}?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            container_app_name,
+            function_name,
+            cls.api_version)
+
+        r = send_raw_request(cmd.cli_ctx, "GET", request_url)
+        return r.json()
+
+
 class DaprComponentResiliencyPreviewClient():
     api_version = PREVIEW_API_VERSION
 

src/containerapp/azext_containerapp/_help.py

@@ -179,6 +179,39 @@
           az containerapp replica count -n my-containerapp -g MyResourceGroup
 """
 
+helps['containerapp functions'] = """
+    type: group
+    short-summary: Commands to view functions in a container app.
+"""
+
+helps['containerapp functions list'] = """
+    type: command
+    short-summary: List all functions in a container app or a specific revision.
+    long-summary: |
+        When revision name is not provided, lists functions from the active revision. If there are multiple active revisions, the API will return an error requiring you to specify which revision.
+    examples:
+    - name: List all functions in a container app (single active revision mode)
+      text: |
+          az containerapp functions list -n my-containerapp -g MyResourceGroup
+    - name: List all functions for a specific revision
+      text: |
+          az containerapp functions list -n my-containerapp -g MyResourceGroup --revision-name MyRevision
+"""
+
+helps['containerapp functions show'] = """
+    type: command
+    short-summary: Show details of a specific function in a container app.
+    long-summary: |
+        When revision name is not provided, shows function from the active revision. If there are multiple active revisions, the API will return an error requiring you to specify which revision.
+    examples:
+    - name: Show details of a function in a container app (single active revision mode)
+      text: |
+          az containerapp functions show -n my-containerapp -g MyResourceGroup --function-name MyFunction
+    - name: Show details of a function for a specific revision
+      text: |
+          az containerapp functions show -n my-containerapp -g MyResourceGroup --function-name MyFunction --revision-name MyRevision
+"""
+
 # Environment Commands
 helps['containerapp env'] = """
     type: group

src/containerapp/azext_containerapp/_params.py

@@ -529,3 +529,12 @@ 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 functions list') 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_name', options_list=['--revision-name', '-r'], help="The name of the revision to list functions from. If not provided, lists functions from the active revision (only works if there's a single active revision).")
+
+    with self.argument_context('containerapp functions show') as c:
+        c.argument('function_name', options_list=['--function-name', '-f'], help="The name of the function to show details for.")
+        c.argument('revision_name', options_list=['--revision-name', '-r'], help="The name of the revision to get the function from. If not provided, gets function from the active revision (only works if there's a single active revision).")

src/containerapp/azext_containerapp/commands.py

@@ -32,6 +32,10 @@ def load_command_table(self, args):
         g.custom_command('list', 'list_replicas')
         g.custom_command('count', 'count_replicas', is_preview=True)
 
+    with self.command_group('containerapp functions', is_preview=True) as g:
+        g.custom_command('list', 'list_containerapp_functions')
+        g.custom_show_command('show', 'show_containerapp_function')
+
     with self.command_group('containerapp env') as g:
         g.custom_show_command('show', 'show_managed_environment')
         g.custom_command('list', 'list_managed_environments')

src/containerapp/azext_containerapp/containerapp_functions_decorator.py

@@ -0,0 +1,113 @@
+# 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, too-many-public-methods, too-many-boolean-expressions, logging-fstring-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, ResourceNotFoundError)
+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_container_app_name(self):
+        return self.get_param("container_app_name")
+
+    def get_argument_revision_name(self):
+        return self.get_param("revision_name")
+
+    def get_argument_function_name(self):
+        return self.get_param("function_name")
+
+    def set_argument_container_app_name(self, container_app_name):
+        self.set_param("container_app_name", container_app_name)
+
+    def set_argument_revision_name(self, revision_name):
+        self.set_param("revision_name", revision_name)
+
+    def set_argument_function_name(self, function_name):
+        self.set_param("function_name", function_name)
+
+
+class ContainerAppFunctionsListDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for listing functions"""
+    
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def list(self):
+        """List functions for a container app or revision"""
+        try:
+            revision_name = self.get_argument_revision_name()
+            container_app_name = self.get_argument_container_app_name()
+            resource_group_name = self.get_argument_resource_group_name()
+
+            if revision_name:
+                # List functions for a specific revision
+                return self.client.list_functions_by_revision(
+                    cmd=self.cmd,
+                    resource_group_name=resource_group_name,
+                    container_app_name=container_app_name,
+                    revision_name=revision_name
+                )
+            else:
+                # List functions for the entire container app
+                return self.client.list_functions(
+                    cmd=self.cmd,
+                    resource_group_name=resource_group_name,
+                    container_app_name=container_app_name
+                )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionsShowDecorator(ContainerAppFunctionsDecorator):
+    """Decorator for showing a specific function"""
+    
+    def __init__(self, cmd: AzCliCommand, client: Any, raw_parameters: Dict, models: str):
+        super().__init__(cmd, client, raw_parameters, models)
+
+    def show(self):
+        """Show details of a specific function"""
+        try:
+            revision_name = self.get_argument_revision_name()
+            container_app_name = self.get_argument_container_app_name()
+            function_name = self.get_argument_function_name()
+            resource_group_name = self.get_argument_resource_group_name()
+
+            if not function_name:
+                raise ValidationError("Function name is required.")
+
+            if revision_name:
+                # Get function for a specific revision
+                return self.client.get_function_by_revision(
+                    cmd=self.cmd,
+                    resource_group_name=resource_group_name,
+                    container_app_name=container_app_name,
+                    revision_name=revision_name,
+                    function_name=function_name
+                )
+            else:
+                # Get function for the entire container app
+                return self.client.get_function(
+                    cmd=self.cmd,
+                    resource_group_name=resource_group_name,
+                    container_app_name=container_app_name,
+                    function_name=function_name
+                )
+        except Exception as e:
+            handle_raw_exception(e)

src/containerapp/azext_containerapp/custom.py

@@ -93,6 +93,10 @@
 from .containerapp_session_code_interpreter_decorator import SessionCodeInterpreterCommandsPreviewDecorator
 from .containerapp_job_registry_decorator import ContainerAppJobRegistryPreviewSetDecorator
 from .containerapp_env_maintenance_config_decorator import ContainerAppEnvMaintenanceConfigPreviewDecorator
+from .containerapp_functions_decorator import (
+    ContainerAppFunctionsListDecorator,
+    ContainerAppFunctionsShowDecorator
+)
 from .dotnet_component_decorator import DotNetComponentDecorator
 from ._client_factory import handle_raw_exception, handle_non_404_status_code_exception
 from ._clients import (
@@ -114,7 +118,8 @@
     DotNetComponentPreviewClient,
     MaintenanceConfigPreviewClient,
     HttpRouteConfigPreviewClient,
-    LabelHistoryPreviewClient
+    LabelHistoryPreviewClient,
+    ContainerAppFunctionsPreviewClient
 )
 from ._dev_service_utils import DevServiceUtils
 from ._models import (
@@ -3937,3 +3942,37 @@ def remove_environment_premium_ingress(cmd, name, resource_group_name, no_wait=F
 
     except Exception as e:
         handle_raw_exception(e)
+
+
+# Container App Functions commands
+def list_containerapp_functions(cmd, resource_group_name, name, revision_name=None):
+    """List functions for a container app or specific revision"""
+    containerapp_functions_list_decorator = ContainerAppFunctionsListDecorator(
+        cmd=cmd,
+        client=ContainerAppFunctionsPreviewClient,
+        raw_parameters={
+            'resource_group_name': resource_group_name,
+            'container_app_name': name,
+            'revision_name': revision_name
+        },
+        models=CONTAINER_APPS_SDK_MODELS
+    )
+
+    return containerapp_functions_list_decorator.list()
+
+
+def show_containerapp_function(cmd, resource_group_name, name, function_name, revision_name=None):
+    """Show details of a specific function for a container app or revision"""
+    containerapp_functions_show_decorator = ContainerAppFunctionsShowDecorator(
+        cmd=cmd,
+        client=ContainerAppFunctionsPreviewClient,
+        raw_parameters={
+            'resource_group_name': resource_group_name,
+            'container_app_name': name,
+            'function_name': function_name,
+            'revision_name': revision_name
+        },
+        models=CONTAINER_APPS_SDK_MODELS
+    )
+
+    return containerapp_functions_show_decorator.show()