adding all the missing files

Author: khushishah513

src/containerapp/azext_containerapp/containerapp_debug_command_decorator.py

@@ -0,0 +1,89 @@
+# 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
+import urllib
+from azure.cli.core.azclierror import ValidationError
+from azure.cli.command_modules.containerapp.base_resource import BaseResource
+from azure.cli.core.commands.client_factory import get_subscription_id
+from azure.cli.core.util import send_raw_request
+from ._transformers import transform_debug_command_output
+
+from ._validators import validate_basic_arguments
+
+logger = get_logger(__name__)
+
+
+class ContainerAppDebugCommandDecorator(BaseResource):
+    """Base decorator for Container App Debug Command Operations"""
+
+    def get_argument_resource_group_name(self):
+        return self.get_param('resource_group_name')
+
+    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_replica_name(self):
+        return self.get_param("replica_name")
+
+    def get_argument_container_name(self):
+        return self.get_param("container_name")
+
+    def get_argument_command(self):
+        return self.get_param("command")
+
+    def validate_arguments(self):
+        validate_basic_arguments(
+            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(),
+            replica_name=self.get_argument_replica_name(),
+            container_name=self.get_argument_container_name(),
+            command=self.get_argument_command()
+        )
+
+    def _get_logstream_endpoint(self, cmd, resource_group_name, container_app_name, revision_name, replica_name, container_name):
+        """Get the logstream endpoint for the specified container in the replica"""
+        containers = self.client.get_replica(cmd,
+                                             resource_group_name,
+                                             container_app_name, revision_name, replica_name)["properties"]["containers"]
+        container_info = [c for c in containers if c["name"] == container_name]
+        if not container_info:
+            raise ValidationError(f"Error retrieving container in revision '{revision_name}' in the container app '{container_app_name}'.")
+        return container_info[0]["logStreamEndpoint"]
+
+    def _get_url(self, cmd, resource_group_name, container_app_name, revision_name, replica_name, container_name, command):
+        """Get the debug url for the specified container in the replica"""
+        base_url = self._get_logstream_endpoint(cmd, resource_group_name, container_app_name, revision_name, replica_name, container_name)
+        proxy_api_url = base_url[:base_url.index("/subscriptions/")]
+        sub = get_subscription_id(cmd.cli_ctx)
+        encoded_cmd = urllib.parse.quote_plus(command)
+        debug_url = (f"{proxy_api_url}/subscriptions/{sub}/resourceGroups/{resource_group_name}/containerApps/{container_app_name}"
+                     f"/revisions/{revision_name}/replicas/{replica_name}/debug"
+                     f"?targetContainer={container_name}&command={encoded_cmd}")
+        return debug_url
+
+    def _get_auth_token(self, cmd, resource_group_name, container_app_name):
+        token_response = self.client.get_auth_token(cmd, resource_group_name, container_app_name)
+        return token_response["properties"]["token"]
+
+    def execute_Command(self, cmd):
+        """Execute the command in the specified container in the replica"""
+        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()
+        replica_name = self.get_argument_replica_name()
+        container_name = self.get_argument_container_name()
+        command = self.get_argument_command()
+        url = self._get_url(cmd, resource_group_name, container_app_name, revision_name, replica_name, container_name, command)
+        token = self._get_auth_token(cmd, resource_group_name, container_app_name)
+        headers = [f"Authorization=Bearer {token}"]
+        r = send_raw_request(cmd.cli_ctx, "GET", url, headers=headers)
+        return transform_debug_command_output(r.json())

src/containerapp/azext_containerapp/containerapp_function_keys_decorator.py

@@ -0,0 +1,184 @@
+# 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 azure.cli.core.azclierror import ValidationError
+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 ._utils import get_random_replica
+
+logger = get_logger(__name__)
+
+
+class ContainerAppFunctionKeysDecorator(BaseResource):
+    """Base decorator for Container App Function Keys operations"""
+
+    def get_argument_function_name(self):
+        return self.get_param("function_name")
+
+    def get_argument_revision(self):
+        return self.get_param("revision_name")
+
+    def get_argument_key_type(self):
+        return self.get_param("key_type")
+
+    def get_argument_name(self):
+        return self.get_param("container_app_name")
+
+    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 validate_common_arguments(self):
+        """Validate common arguments required for all function key operations"""
+        resource_group_name = self.get_argument_resource_group_name()
+        name = self.get_argument_name()
+        revision_name = self.get_argument_revision()
+        key_type = self.get_argument_key_type()
+
+        # Validate basic arguments
+        validate_basic_arguments(
+            resource_group_name=resource_group_name,
+            container_app_name=name,
+            key_type=key_type
+        )
+
+        # Validate that the Container App has kind 'functionapp'
+        validate_functionapp_kind(
+            cmd=self.cmd,
+            resource_group_name=resource_group_name,
+            container_app_name=name
+        )
+
+        # Validate revision and get the appropriate revision name
+        revision_name = validate_revision_and_get_name(
+            cmd=self.cmd,
+            resource_group_name=resource_group_name,
+            container_app_name=name,
+            provided_revision_name=revision_name
+        )
+
+        # Get a random replica for the revision
+        replica_name, container_name = get_random_replica(
+            cmd=self.cmd,
+            resource_group_name=resource_group_name,
+            container_app_name=name,
+            revision_name=revision_name
+        )
+
+        return resource_group_name, name, revision_name, key_type, replica_name, container_name
+
+    def validate_function_name_requirement(self, key_type):
+        """Validate function name is provided when required for functionKey type"""
+        function_name = self.get_argument_function_name()
+
+        if key_type == "functionKey" and not function_name:
+            raise ValidationError("Function name is required when key-type is 'functionKey'.")
+
+        return function_name
+
+
+class ContainerAppFunctionKeysShowDecorator(ContainerAppFunctionKeysDecorator):
+    """Decorator for showing specific function keys"""
+
+    def validate_show_arguments(self):
+        """Validate arguments required for showing function keys"""
+        resource_group_name, name, revision_name, key_type, replica_name, container_name = self.validate_common_arguments()
+        key_name = self.get_argument_key_name()
+        function_name = self.validate_function_name_requirement(key_type)
+
+        if not key_name:
+            raise ValidationError("Key name is required.")
+
+        return resource_group_name, name, revision_name, key_type, key_name, function_name, replica_name, container_name
+
+    def show_keys(self):
+        """Show specific key"""
+        try:
+            resource_group_name, name, revision_name, key_type, key_name, function_name, replica_name, container_name = self.validate_show_arguments()
+
+            return self.client.show_function_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                key_type=key_type,
+                key_name=key_name,
+                function_name=function_name,
+                revision_name=revision_name,
+                replica_name=replica_name,
+                container_name=container_name
+            )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionKeysListDecorator(ContainerAppFunctionKeysDecorator):
+    """Decorator for listing function keys"""
+
+    def validate_list_arguments(self):
+        """Validate arguments required for listing function keys"""
+        resource_group_name, name, revision_name, key_type, replica_name, container_name = self.validate_common_arguments()
+        function_name = self.validate_function_name_requirement(key_type)
+
+        return resource_group_name, name, revision_name, key_type, function_name, replica_name, container_name
+
+    def list_keys(self):
+        """List keys based on key type"""
+        try:
+            resource_group_name, name, revision_name, key_type, function_name, replica_name, container_name = self.validate_list_arguments()
+
+            return self.client.list_function_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                key_type=key_type,
+                function_name=function_name,
+                revision_name=revision_name,
+                replica_name=replica_name,
+                container_name=container_name
+            )
+        except Exception as e:
+            handle_raw_exception(e)
+
+
+class ContainerAppFunctionKeysSetDecorator(ContainerAppFunctionKeysDecorator):
+    """Decorator for creating/updating function keys"""
+
+    def validate_set_arguments(self):
+        """Validate arguments required for setting/updating function keys"""
+        resource_group_name, name, revision_name, key_type, replica_name, container_name = self.validate_common_arguments()
+        key_name = self.get_argument_key_name()
+        key_value = self.get_argument_key_value()
+        function_name = self.validate_function_name_requirement(key_type)
+
+        if not key_name:
+            raise ValidationError("Key name is required.")
+
+        return resource_group_name, name, revision_name, key_type, key_name, key_value, function_name, replica_name, container_name
+
+    def set_keys(self):
+        """Create/Update keys based on key type"""
+        try:
+            resource_group_name, name, revision_name, key_type, key_name, key_value, function_name, replica_name, container_name = self.validate_set_arguments()
+
+            return self.client.set_function_keys(
+                cmd=self.cmd,
+                resource_group_name=resource_group_name,
+                name=name,
+                key_type=key_type,
+                key_name=key_name,
+                key_value=key_value,
+                function_name=function_name,
+                revision_name=revision_name,
+                replica_name=replica_name,
+                container_name=container_name
+            )
+        except Exception as e:
+            handle_raw_exception(e)