Merge pull request #20 from khushishah513/function-comments

resolving comments on PR

Author: khushishah513

src/containerapp/azext_containerapp/_clients.py

@@ -305,7 +305,6 @@ def list(cls, cmd, resource_group_name, container_app_name):
 
 class ContainerAppFunctionsPreviewClient():
     api_version = "2025-10-02-preview"
-    APP_INSIGHTS_API_VERSION = "2018-04-20"
 
     @classmethod
     def list_functions_by_revision(cls, cmd, resource_group_name, container_app_name, revision_name):
@@ -379,114 +378,9 @@ def get_function(cls, cmd, resource_group_name, container_app_name, function_nam
             raise CLIError(f"Error retrieving function '{function_name}' for container app '{container_app_name}'.")
         return r.json()
 
-    @classmethod
-    def get_function_invocation_summary(cls, cmd, resource_group_name, container_app_name, revision_name, function_name, timespan="30d"):
-        # Fetch the app insights resource app id
-        app_id = cls._get_app_insights_id(cmd, resource_group_name, container_app_name, revision_name)
-
-        # Use application insights query to get function invocations summary
-        invocation_summary_query = (
-            f"requests | extend functionNameFromCustomDimension = tostring(customDimensions['faas.name']) "
-            f"| where timestamp >= ago({timespan}) "
-            f"| where cloud_RoleName =~ '{container_app_name}' "
-            f"| where cloud_RoleInstance contains '{revision_name}' "
-            f"| where operation_Name =~ '{function_name}' or functionNameFromCustomDimension =~ '{function_name}' "
-            f"| summarize SuccessCount = coalesce(countif(success == true), 0), ErrorCount = coalesce(countif(success == false), 0)"
-        )
-
-        try:
-            result = cls._execute_app_insights_query(cmd, app_id, invocation_summary_query, "getLast30DaySummary")
-            return result
-        except Exception as ex:
-            raise CLIError(f"Error retrieving function invocation summary: {str(ex)}")
-
-    @classmethod
-    def get_function_invocation_traces(cls, cmd, resource_group_name, container_app_name, revision_name, function_name, timespan="30d", limit=20):
-        # Fetch the app insights resource app id
-        app_id = cls._get_app_insights_id(cmd, resource_group_name, container_app_name, revision_name)
-
-        # Use application insights query to get function invocations traces
-        invocation_traces_query = (
-            f"requests | extend functionNameFromCustomDimension = tostring(customDimensions['faas.name']) "
-            f"| project timestamp, id, operation_Name, success, resultCode, duration, operation_Id, functionNameFromCustomDimension, "
-            f"cloud_RoleName, cloud_RoleInstance, invocationId=coalesce(tostring(customDimensions['InvocationId']), tostring(customDimensions['faas.invocation_id'])) "
-            f"| where timestamp > ago({timespan}) "
-            f"| where cloud_RoleName =~ '{container_app_name}' "
-            f"| where cloud_RoleInstance contains '{revision_name}' "
-            f"| where operation_Name =~ '{function_name}' or functionNameFromCustomDimension =~ '{function_name}' "
-            f"| order by timestamp desc | take {limit} "
-            f"| project timestamp, success, resultCode, durationInMilliSeconds=duration, invocationId, operationId=operation_Id, operationName=operation_Name, functionNameFromCustomDimension "
-        )
-
-        try:
-            result = cls._execute_app_insights_query(cmd, app_id, invocation_traces_query, "getInvocationTraces")
-            return result
-        except Exception as ex:
-            raise CLIError(f"Error retrieving function invocation traces: {str(ex)}")
-
-    @classmethod
-    def _get_app_insights_id(cls, cmd, resource_group_name, container_app_name, revision_name):
-        # Fetch the revision details using the container app client
-        revision = ContainerAppPreviewClient.show_revision(cmd, resource_group_name, container_app_name, revision_name)
-        # Extract the list of environment variables from the revision's properties
-        env_vars = []
-        if revision and "properties" in revision and "template" in revision["properties"]:
-            containers = revision["properties"]["template"].get("containers", [])
-            for container in containers:
-                env_vars.extend(container.get("env", []))
-
-        # Check for APPLICATIONINSIGHTS_CONNECTION_STRING
-        ai_conn_str = None
-        for env in env_vars:
-            if env.get("name") == "APPLICATIONINSIGHTS_CONNECTION_STRING":
-                ai_conn_str = env.get("value")
-                break
-
-        if not ai_conn_str:
-            raise CLIError(f"Required application setting APPLICATIONINSIGHTS_CONNECTION_STRING not present in the containerapp '{container_app_name}'.")
-
-        # Extract ApplicationId from the connection string
-        app_id = None
-        parts = ai_conn_str.split(";")
-        for part in parts:
-            if part.startswith("ApplicationId="):
-                app_id = part.split("=", 1)[1]
-                break
-
-        if not app_id:
-            raise CLIError(f"ApplicationId not found in APPLICATIONINSIGHTS_CONNECTION_STRING for containerapp '{container_app_name}'.")
-        return app_id
-
-    @classmethod
-    def _execute_app_insights_query(cls, cmd, app_id, query, queryType, timespan="30D"):
-
-        # Application Insights REST API endpoint
-        api_endpoint = "https://api.applicationinsights.io"
-        url = f"{api_endpoint}/v1/apps/{app_id}/query?api-version={cls.APP_INSIGHTS_API_VERSION}&queryType={queryType}"
-
-        # Prepare the request body
-        body = {
-            "query": query,
-            "timespan": f"P{timespan}"
-        }
-
-        # Execute the query using Azure CLI's send_raw_request
-        response = send_raw_request(
-            cmd.cli_ctx,
-            "POST",
-            url,
-            body=json.dumps(body),
-            headers=["Content-Type=application/json"]
-        )
-
-        result = response.json()
-        if isinstance(result, dict) and 'error' in result:
-            raise CLIError(f"Error retrieving invocations details: {result['error']}")
-        return result
-
     @classmethod
     def show_function_keys(cls, cmd, resource_group_name, name, key_type, key_name, function_name=None, revision_name=None, replica_name=None, container_name=None):
-        from .custom import containerapp_debug
+        from ._utils import execute_function_admin_command
 
         command_fmt = ""
         if key_type != "functionKey":
@@ -496,22 +390,22 @@ def show_function_keys(cls, cmd, resource_group_name, name, key_type, key_name,
             command_fmt = "/bin/azure-functions-admin keys show --key-type {} --key-name {} --function-name {}"
             command = command_fmt.format(key_type, key_name, function_name)
 
-        r = containerapp_debug(
+        r = execute_function_admin_command(
             cmd=cmd,
             resource_group_name=resource_group_name,
             name=name,
-            container=container_name,
-            revision=revision_name,
-            replica=replica_name,
-            debug_command=command
+            command=command,
+            revision_name=revision_name,
+            replica_name=replica_name,
+            container_name=container_name
         )
         if not r:
             raise CLIError(f"Error retrieving function key '{key_name}' of type '{key_type}'.")
         return r
 
     @classmethod
     def list_function_keys(cls, cmd, resource_group_name, name, key_type, function_name=None, revision_name=None, replica_name=None, container_name=None):
-        from .custom import containerapp_debug
+        from ._utils import execute_function_admin_command
 
         command_fmt = ""
         if key_type != "functionKey":
@@ -521,14 +415,14 @@ def list_function_keys(cls, cmd, resource_group_name, name, key_type, function_n
             command_fmt = "/bin/azure-functions-admin keys list --key-type {} --function-name {}"
             command = command_fmt.format(key_type, function_name)
 
-        r = containerapp_debug(
+        r = execute_function_admin_command(
             cmd=cmd,
             resource_group_name=resource_group_name,
             name=name,
-            container=container_name,
-            revision=revision_name,
-            replica=replica_name,
-            debug_command=command
+            command=command,
+            revision_name=revision_name,
+            replica_name=replica_name,
+            container_name=container_name
         )
         if not r:
             raise CLIError(f"Error retrieving function keys of type '{key_type}'.")
@@ -537,7 +431,7 @@ def list_function_keys(cls, cmd, resource_group_name, name, key_type, function_n
     @classmethod
     def set_function_keys(cls, cmd, resource_group_name, name, key_type, key_name, key_value, function_name=None, revision_name=None, replica_name=None, container_name=None):
         """Set/Update function keys based on key type"""
-        from .custom import containerapp_debug
+        from ._utils import execute_function_admin_command
 
         command_fmt = ""
         if key_type != "functionKey":
@@ -550,14 +444,14 @@ def set_function_keys(cls, cmd, resource_group_name, name, key_type, key_name, k
         if key_value is not None:
             command += " --key-value {}".format(key_value)
 
-        r = containerapp_debug(
+        r = execute_function_admin_command(
             cmd=cmd,
             resource_group_name=resource_group_name,
             name=name,
-            container=container_name,
-            revision=revision_name,
-            replica=replica_name,
-            debug_command=command
+            command=command,
+            revision_name=revision_name,
+            replica_name=replica_name,
+            container_name=container_name
         )
         if not r:
             raise CLIError(f"Error setting function key '{key_name}' of type '{key_type}'.")

src/containerapp/azext_containerapp/_utils.py

@@ -875,3 +875,17 @@ def get_random_replica(cmd, resource_group_name, container_app_name, revision_na
         raise CLIError(f"Could not determine replica name for revision '{revision_name}' of container app '{container_app_name}'.")
 
     return replica_name, container_name
+
+
+def execute_function_admin_command(cmd, resource_group_name, name, command, revision_name=None, replica_name=None, container_name=None):
+    from .custom import containerapp_debug
+    
+    return containerapp_debug(
+        cmd=cmd,
+        resource_group_name=resource_group_name,
+        name=name,
+        container=container_name,
+        revision=revision_name,
+        replica=replica_name,
+        debug_command=command
+    )

src/containerapp/azext_containerapp/containerapp_functions_decorator.py

@@ -5,14 +5,17 @@
 # --------------------------------------------------------------------------------------------
 # pylint: disable=line-too-long, broad-except, logging-format-interpolation, too-many-public-methods, too-many-boolean-expressions, logging-fstring-interpolation
 
+import json
 from knack.log import get_logger
 
-from azure.cli.core.azclierror import ValidationError
+from azure.cli.core.azclierror import ValidationError, CLIError
+from azure.cli.core.util import send_raw_request
 from azure.cli.command_modules.containerapp.base_resource import BaseResource
 
 from ._client_factory import handle_raw_exception
 from ._validators import validate_basic_arguments, validate_revision_and_get_name, validate_functionapp_kind
 from ._transformers import process_app_insights_response
+from ._clients import ContainerAppPreviewClient
 
 logger = get_logger(__name__)
 
@@ -165,6 +168,8 @@ def show(self):
 class ContainerAppFunctionInvocationsDecorator(ContainerAppFunctionsDecorator):
     """Decorator for showing function invocation"""
 
+    APP_INSIGHTS_API_VERSION = "2018-04-20"
+
     def validate_arguments(self):
         """Validate arguments required for function invocation operations"""
         validate_basic_arguments(
@@ -190,21 +195,92 @@ def validate_arguments(self):
         self.set_argument_revision_name(revision_name)
         self.validate_function_name_requirement()
 
+    def _get_app_insights_id(self, resource_group_name, container_app_name, revision_name):
+        # Fetch the revision details using the container app client
+        revision = ContainerAppPreviewClient.show_revision(self.cmd, resource_group_name, container_app_name, revision_name)
+        # Extract the list of environment variables from the revision's properties
+        env_vars = []
+        if revision and "properties" in revision and "template" in revision["properties"]:
+            containers = revision["properties"]["template"].get("containers", [])
+            for container in containers:
+                env_vars.extend(container.get("env", []))
+
+        # Check for APPLICATIONINSIGHTS_CONNECTION_STRING
+        ai_conn_str = None
+        for env in env_vars:
+            if env.get("name") == "APPLICATIONINSIGHTS_CONNECTION_STRING":
+                ai_conn_str = env.get("value")
+                break
+
+        if not ai_conn_str:
+            raise CLIError(f"Required application setting APPLICATIONINSIGHTS_CONNECTION_STRING not present in the containerapp '{container_app_name}'.")
+
+        # Extract ApplicationId from the connection string
+        app_id = None
+        parts = ai_conn_str.split(";")
+        for part in parts:
+            if part.startswith("ApplicationId="):
+                app_id = part.split("=", 1)[1]
+                break
+
+        if not app_id:
+            raise CLIError(f"ApplicationId not found in APPLICATIONINSIGHTS_CONNECTION_STRING for containerapp '{container_app_name}'.")
+        return app_id
+
+    def _execute_app_insights_query(self, app_id, query, query_type, timespan="30D"):
+        # Application Insights REST API endpoint
+        api_endpoint = "https://api.applicationinsights.io"
+        url = f"{api_endpoint}/v1/apps/{app_id}/query?api-version={self.APP_INSIGHTS_API_VERSION}&queryType={query_type}"
+
+        # Prepare the request body
+        body = {
+            "query": query,
+            "timespan": f"P{timespan}"
+        }
+
+        # Execute the query using Azure CLI's send_raw_request
+        response = send_raw_request(
+            self.cmd.cli_ctx,
+            "POST",
+            url,
+            body=json.dumps(body),
+            headers=["Content-Type=application/json"]
+        )
+
+        result = response.json()
+        if isinstance(result, dict) and 'error' in result:
+            raise CLIError(f"Error retrieving invocations details: {result['error']}")
+        return result
+
     def get_summary(self):
         """Get function invocation summary using the client"""
         try:
             self.validate_arguments()
 
-            response = self.client.get_function_invocation_summary(
-                cmd=self.cmd,
-                resource_group_name=self.get_argument_resource_group_name(),
-                container_app_name=self.get_argument_container_app_name(),
-                revision_name=self.get_argument_revision_name(),
-                function_name=self.get_argument_function_name(),
-                timespan=self.get_argument_timespan() or "30d"
+            # Get arguments
+            cmd = self.cmd
+            resource_group_name = self.get_argument_resource_group_name()
+            container_app_name = self.get_argument_container_app_name()
+            revision_name = self.get_argument_revision_name()
+            function_name = self.get_argument_function_name()
+            timespan = self.get_argument_timespan() or "30d"
+
+            # Fetch the app insights resource app id
+            app_id = self._get_app_insights_id(resource_group_name, container_app_name, revision_name)
+
+            # Use application insights query to get function invocations summary
+            invocation_summary_query = (
+                f"requests | extend functionNameFromCustomDimension = tostring(customDimensions['faas.name']) "
+                f"| where timestamp >= ago({timespan}) "
+                f"| where cloud_RoleName =~ '{container_app_name}' "
+                f"| where cloud_RoleInstance contains '{revision_name}' "
+                f"| where operation_Name =~ '{function_name}' or functionNameFromCustomDimension =~ '{function_name}' "
+                f"| summarize SuccessCount = coalesce(countif(success == true), 0), ErrorCount = coalesce(countif(success == false), 0)"
             )
 
-            return process_app_insights_response(response)
+            result = self._execute_app_insights_query(app_id, invocation_summary_query, "getLast30DaySummary")
+
+            return process_app_insights_response(result)
         except Exception as e:
             handle_raw_exception(e)
 
@@ -214,23 +290,32 @@ def get_traces(self):
             self.validate_arguments()
 
             # Get all arguments
-            resource_group = self.get_argument_resource_group_name()
-            container_app = self.get_argument_container_app_name()
-            revision = self.get_argument_revision_name()
-            function = self.get_argument_function_name()
-            timespan = self.get_argument_timespan()
-            limit = self.get_argument_limit()
-
-            response = self.client.get_function_invocation_traces(
-                cmd=self.cmd,
-                resource_group_name=resource_group,
-                container_app_name=container_app,
-                revision_name=revision,
-                function_name=function,
-                timespan=timespan,
-                limit=limit
+            cmd = self.cmd
+            resource_group_name = self.get_argument_resource_group_name()
+            container_app_name = self.get_argument_container_app_name()
+            revision_name = self.get_argument_revision_name()
+            function_name = self.get_argument_function_name()
+            timespan = self.get_argument_timespan() or "30d"
+            limit = self.get_argument_limit() or 20
+
+            # Fetch the app insights resource app id
+            app_id = self._get_app_insights_id(resource_group_name, container_app_name, revision_name)
+
+            # Use application insights query to get function invocations traces
+            invocation_traces_query = (
+                f"requests | extend functionNameFromCustomDimension = tostring(customDimensions['faas.name']) "
+                f"| project timestamp, id, operation_Name, success, resultCode, duration, operation_Id, functionNameFromCustomDimension, "
+                f"cloud_RoleName, cloud_RoleInstance, invocationId=coalesce(tostring(customDimensions['InvocationId']), tostring(customDimensions['faas.invocation_id'])) "
+                f"| where timestamp > ago({timespan}) "
+                f"| where cloud_RoleName =~ '{container_app_name}' "
+                f"| where cloud_RoleInstance contains '{revision_name}' "
+                f"| where operation_Name =~ '{function_name}' or functionNameFromCustomDimension =~ '{function_name}' "
+                f"| order by timestamp desc | take {limit} "
+                f"| project timestamp, success, resultCode, durationInMilliSeconds=duration, invocationId, operationId=operation_Id, operationName=operation_Name, functionNameFromCustomDimension "
             )
 
-            return process_app_insights_response(response)
+            result = self._execute_app_insights_query(app_id, invocation_traces_query, "getInvocationTraces")
+
+            return process_app_insights_response(result)
         except Exception as e:
             handle_raw_exception(e)