content: #  Copyright (c) ZenML GmbH 2023. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """Azure Service Connector."""
  
  import datetime
  import json
  import logging
  import re
  import subprocess
  from typing import Any, Dict, List, Optional, Tuple
  from uuid import UUID
  
  import requests
  import yaml
  from azure.core.credentials import AccessToken, TokenCredential
  from azure.core.exceptions import AzureError
  from azure.identity import ClientSecretCredential, DefaultAzureCredential
  from azure.mgmt.containerregistry import ContainerRegistryManagementClient
  from azure.mgmt.containerservice import ContainerServiceClient
  from azure.mgmt.resource import SubscriptionClient
  from azure.mgmt.storage import StorageManagementClient
  from azure.storage.blob import BlobServiceClient
  from pydantic import Field
  
  from zenml.constants import (
      DOCKER_REGISTRY_RESOURCE_TYPE,
      KUBERNETES_CLUSTER_RESOURCE_TYPE,
  )
  from zenml.exceptions import AuthorizationException
  from zenml.integrations.azure import (
      AZURE_CONNECTOR_TYPE,
      AZURE_RESOURCE_TYPE,
      BLOB_RESOURCE_TYPE,
  )
  from zenml.logger import get_logger
  from zenml.models import (
      AuthenticationMethodModel,
      ResourceTypeModel,
      ServiceConnectorTypeModel,
  )
  from zenml.service_connectors.docker_service_connector import (
      DockerAuthenticationMethods,
      DockerConfiguration,
      DockerServiceConnector,
  )
  from zenml.service_connectors.service_connector import (
      AuthenticationConfig,
      ServiceConnector,
  )
  from zenml.utils.enum_utils import StrEnum
  from zenml.utils.secret_utils import PlainSerializedSecretStr
  from zenml.utils.time_utils import to_local_tz, to_utc_timezone, utc_now
  
  # Configure the logging level for azure.identity
  logging.getLogger("azure.identity").setLevel(logging.WARNING)
  
  logger = get_logger(__name__)
  
  
  AZURE_MANAGEMENT_TOKEN_SCOPE = "https://management.azure.com/.default"
  AZURE_SESSION_TOKEN_DEFAULT_EXPIRATION_TIME = 60 * 60  # 1 hour
  AZURE_SESSION_EXPIRATION_BUFFER = 15  # 15 minutes
  AZURE_ACR_OAUTH_SCOPE = "repository:*:*"
  
  
  class AzureBaseConfig(AuthenticationConfig):
      """Azure base configuration."""
  
      subscription_id: Optional[UUID] = Field(
          default=None,
          title="Azure Subscription ID",
          description="The subscription ID of the Azure account. If not "
          "specified, ZenML will attempt to retrieve the subscription ID from "
          "Azure using the configured credentials.",
      )
      tenant_id: Optional[UUID] = Field(
          default=None,
          title="Azure Tenant ID",
          description="The tenant ID of the Azure account. If not specified, "
          "ZenML will attempt to retrieve the tenant from Azure using the "
          "configured credentials.",
      )
      resource_group: Optional[str] = Field(
          default=None,
          title="Azure Resource Group",
          description="A resource group may be used to restrict the scope of "
          "Azure resources like AKS clusters and ACR repositories. If not "
          "specified, ZenML will retrieve resources from all resource groups "
          "accessible with the configured credentials.",
      )
      storage_account: Optional[str] = Field(
          default=None,
          title="Azure Storage Account",
          description="The name of an Azure storage account may be used to "
          "restrict the scope of Azure Blob storage containers. If not "
          "specified, ZenML will retrieve blob containers from all storage "
          "accounts accessible with the configured credentials.",
      )
  
  
  class AzureClientSecret(AuthenticationConfig):
      """Azure client secret credentials."""
  
      client_secret: PlainSerializedSecretStr = Field(
          title="Service principal client secret",
          description="The client secret of the service principal",
      )
  
  
  class AzureClientConfig(AzureBaseConfig):
      """Azure client configuration."""
  
      client_id: UUID = Field(
          title="Azure Client ID",
          description="The service principal's client ID",
      )
      tenant_id: UUID = Field(
          title="Azure Tenant ID",
          description="ID of the service principal's tenant. Also called its "
          "'directory' ID.",
      )
  
  
  class AzureAccessToken(AuthenticationConfig):
      """Azure access token credentials."""
  
      token: PlainSerializedSecretStr = Field(
          title="Azure Access Token",
          description="The Azure access token to use for authentication",
      )
  
  
  class AzureServicePrincipalConfig(AzureClientConfig, AzureClientSecret):
      """Azure service principal configuration."""
  
  
  class AzureAccessTokenConfig(AzureBaseConfig, AzureAccessToken):
      """Azure token configuration."""
  
  
  class AzureAuthenticationMethods(StrEnum):
      """Azure Authentication methods."""
  
      IMPLICIT = "implicit"
      SERVICE_PRINCIPAL = "service-principal"
      ACCESS_TOKEN = "access-token"
  
  
  class ZenMLAzureTokenCredential(TokenCredential):
      """ZenML Azure token credential.
  
      This class is used to provide a pre-configured token credential to Azure
      clients. Tokens are generated from other Azure credentials and are served
      to Azure clients to authenticate requests.
      """
  
      def __init__(self, token: str, expires_at: datetime.datetime):
          """Initialize ZenML Azure token credential.
  
          Args:
              token: The token to use for authentication
              expires_at: The expiration time of the token
          """
          self.token = token
  
          # Convert the expiration time from UTC to local time
          self.expires_on = int(to_local_tz(expires_at).timestamp())
  
      def get_token(self, *scopes: str, **kwargs: Any) -> Any:
          """Get token.
  
          Args:
              *scopes: Scopes
              **kwargs: Keyword arguments
  
          Returns:
              Token
          """
          return AccessToken(token=self.token, expires_on=self.expires_on)
  
  
  AZURE_SERVICE_CONNECTOR_TYPE_SPEC = ServiceConnectorTypeModel(
      name="Azure Service Connector",
      connector_type=AZURE_CONNECTOR_TYPE,
      description="""
  The ZenML Azure Service Connector facilitates the authentication and access to
  managed Azure services and resources. These encompass a range of resources,
  including blob storage containers, ACR repositories, and AKS clusters.
  
  This connector also supports automatic configuration and detection of
  credentials locally configured through the Azure CLI.
  
  This connector serves as a general means of accessing any Azure service by
  issuing credentials to clients. Additionally, the connector can handle
  specialized authentication for Azure blob storage, Docker and Kubernetes Python
  clients. It also allows for the configuration of local Docker and Kubernetes
  CLIs.
  
  The Azure Service Connector is part of the Azure ZenML integration. You can
  either install the entire integration or use a pypi extra to install it
  independently of the integration:
  
  * `pip install "zenml[connectors-azure]"` installs only prerequisites for the
  Azure Service Connector Type
  * `zenml integration install azure` installs the entire Azure ZenML integration
  
  It is not required to [install and set up the Azure CLI](https://learn.microsoft.com/en-us/cli/azure/authenticate-azure-cli)
  on your local machine to use the Azure Service Connector to link Stack
  Components to Azure resources and services. However, it is recommended to do so
  if you are looking for a quick setup that includes using the auto-configuration
  Service Connector features.
  """,
      supports_auto_configuration=True,
      logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/connectors/azure/azure.png",
      emoji=":regional_indicator_a: ",
      auth_methods=[
          AuthenticationMethodModel(
              name="Azure Implicit Authentication",
              auth_method=AzureAuthenticationMethods.IMPLICIT,
              description="""
  Implicit authentication to Azure services using environment variables, local
  configuration files, workload or managed identities. This authentication method
  doesn't require any credentials to be explicitly configured. It automatically
  discovers and uses credentials from one of the following sources:
  
  - [environment variables](https://learn.microsoft.com/en-us/python/api/overview/azure/identity-readme?view=azure-python#environment-variables)
  - workload identity - if the application is deployed to an Azure Kubernetes
  Service with Managed Identity enabled. This option can only be used when running
  the ZenML server on an AKS cluster.
  - managed identity - if the application is deployed to an Azure host with
  Managed Identity enabled. This option can only be used when running the ZenML client or server on
  an Azure host.
  - Azure CLI - if a user has signed in via [the Azure CLI `az login` command](https://learn.microsoft.com/en-us/cli/azure/authenticate-azure-cli).
  
  This is the quickest and easiest way to authenticate to Azure services. However,
  the results depend on how ZenML is deployed and the environment where it is used
  and is thus not fully reproducible:
  
  - when used with the default local ZenML deployment or a local ZenML server, the
  credentials are the same as those used by the Azure CLI or extracted from local
  environment variables.
  - when connected to a ZenML server, this method only works if the ZenML server
  is deployed in Azure and will use the workload identity attached to the Azure
  resource where the ZenML server is running (e.g. an AKS cluster). The
  permissions of the managed identity may need to be adjusted to allow listing
  and accessing/describing the Azure resources that the connector is configured to
  access.
  
  Note that the discovered credentials inherit the full set of permissions of the
  local Azure CLI configuration, environment variables or remote Azure managed
  identity. Depending on the extent of those permissions, this authentication
  method might not be recommended for production use, as it can lead to accidental
  privilege escalation. Instead, it is recommended to use the Azure service
  principal authentication method to limit the validity and/or permissions of the
  credentials being issued to connector clients.
  """,
              config_class=AzureBaseConfig,
          ),
          AuthenticationMethodModel(
              name="Azure Service Principal",
              auth_method=AzureAuthenticationMethods.SERVICE_PRINCIPAL,
              description="""
  Azure service principal credentials consisting of an Azure client ID and
  client secret. These credentials are used to authenticate clients to Azure
  services.
  
  For this authentication method, the Azure Service Connector requires
  [an Azure service principal to be created](https://learn.microsoft.com/en-us/azure/developer/python/sdk/authentication-on-premises-apps?tabs=azure-portal)
  and a client secret to be generated.
  """,
              config_class=AzureServicePrincipalConfig,
          ),
          AuthenticationMethodModel(
              name="Azure Access Token",
              auth_method=AzureAuthenticationMethods.ACCESS_TOKEN,
              description="""
  Uses temporary Azure access tokens explicitly configured by the user or
  auto-configured from a local environment. This method has the major limitation
  that the user must regularly generate new tokens and update the connector
  configuration as API tokens expire. On the other hand, this method is ideal in
  cases where the connector only needs to be used for a short period of time, such
  as sharing access temporarily with someone else in your team.
  
  This is the authentication method used during auto-configuration, if you have
  [the local Azure CLI set up with credentials](https://learn.microsoft.com/en-us/cli/azure/authenticate-azure-cli).
  The connector will generate an access token from the Azure CLI credentials and
  store it in the connector configuration.
  
  Given that Azure access tokens are scoped to a particular Azure resource and the
  access token generated during auto-configuration is scoped to the Azure
  Management API, this method does not work with Azure blob storage resources. You
  should use the Azure service principal authentication method for blob storage
  resources instead.
  """,
              config_class=AzureAccessTokenConfig,
              default_expiration_seconds=AZURE_SESSION_TOKEN_DEFAULT_EXPIRATION_TIME,
          ),
      ],
      resource_types=[
          ResourceTypeModel(
              name="Generic Azure resource",
              resource_type=AZURE_RESOURCE_TYPE,
              description="""
  Multi-purpose Azure resource type. It allows Stack Components to use the
  connector to connect to any Azure service. When used by Stack Components, they
  are provided generic azure-identity credentials that can be used to create
  Azure python clients for any particular Azure service.
  
  This generic Azure resource type is meant to be used with Stack Components that
  are not represented by other, more specific resource types, like Azure blob
  storage containers, Kubernetes clusters or Docker registries. It should be
  accompanied by a matching set of Azure permissions that allow access to the set
  of remote resources required by the client(s).
  
  The resource name represents the name of the Azure subscription that the
  connector is authorized to access.
  """,
              auth_methods=AzureAuthenticationMethods.values(),
              # Don't request an Azure specific resource instance ID.
              supports_instances=False,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/connectors/azure/azure.png",
              emoji=":regional_indicator_a: ",
          ),
          ResourceTypeModel(
              name="Azure Blob Storage Container",
              resource_type=BLOB_RESOURCE_TYPE,
              description="""
  Allows Stack Components to connect to Azure Blob containers. When used by Stack
  Components, they are provided a pre-configured Azure Blob Storage client.
  
  The configured credentials must have at least the following
  Azure IAM permissions associated with the blob storage account or containers
  that the connector that the connector will be allowed to access:
  
  - allow read and write access to blobs (e.g. the `Storage Blob Data Contributor`
  role)
  - allow listing the storage accounts (e.g. the `Reader and Data
  Access` role). This is only required if a storage account is not configured in
  the connector.
  - allow listing the containers in a storage account (e.g. the `Reader and Data
  Access` role)
  
  If set, the resource name must identify an Azure blob container using one of the following
  formats:
  
  - Azure blob container URI (canonical resource name): `{az|abfs}://{container-name}`
  - Azure blob container name: `{container-name}`
  
  If a storage account is configured in the connector, only blob storage
  containers in that storage account will be accessible. Otherwise, if a resource
  group is configured in the connector, only blob storage containers
  in storage accounts in that resource group will be accessible. Finally, if
  neither a storage account nor a resource group is configured in the connector,
  all blob storage containers in all accessible storage accounts will be
  accessible.
  
  The only Azure authentication methods that work with Azure blob storage resources are the implicit 
  authentication and the service principal authentication method.
  """,
              auth_methods=AzureAuthenticationMethods.values(),
              # Request a blob container to be configured in the
              # connector or provided by the consumer
              supports_instances=True,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/artifact_store/azure.png",
              emoji=":package:",
          ),
          ResourceTypeModel(
              name="AKS Kubernetes cluster",
              resource_type=KUBERNETES_CLUSTER_RESOURCE_TYPE,
              description="""
  Allows Stack Components to access an AKS cluster as a standard Kubernetes
  cluster resource. When used by Stack Components, they are provided a
  pre-authenticated python-kubernetes client instance.
  
  The configured credentials must have at least the following
  Azure IAM permissions associated with the AKS clusters
  that the connector will be allowed to access:
  
  - allow listing the AKS clusters and fetching their credentials (e.g. the
  `Azure Kubernetes Service Cluster Admin Role` role)
  
  If set, the resource name must identify an EKS cluster using one of the
  following formats:
  
  - resource group scoped AKS cluster name (canonical): `[{resource-group}/]{cluster-name}`
  - AKS cluster name: `{cluster-name}`
  
  Given that the AKS cluster name is unique within a resource group, the
  resource group name may be included in the resource name to avoid ambiguity. If
  a resource group is configured in the connector, the resource group name
  in the resource name must match the configured resource group. If no resource
  group is configured in the connector and a resource group name is not included
  in the resource name, the connector will attempt to find the AKS cluster in
  any resource group.
  
  If a resource group is configured in the connector, only AKS clusters in that
  resource group will be accessible.
  """,
              auth_methods=AzureAuthenticationMethods.values(),
              # Request an AKS cluster name to be configured in the
              # connector or provided by the consumer
              supports_instances=True,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/orchestrator/kubernetes.png",
              emoji=":cyclone:",
          ),
          ResourceTypeModel(
              name="ACR container registry",
              resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
              description="""
  Allows Stack Components to access one or more ACR registries as a standard
  Docker registry resource. When used by Stack Components, they are provided a
  pre-authenticated python-docker client instance.
  
  The configured credentials must have at least the following
  Azure IAM permissions associated with the ACR registries
  that the connector that the connector will be allowed to access:
  
  - allow access to pull and push images (e.g. the `AcrPull` and `AcrPush` roles)
  - allow access to list registries (e.g. the `Contributor` role)
  
  If set, the resource name must identify an ACR registry using one of the
  following formats:
  
  - ACR registry URI (canonical resource name): `[https://]{registry-name}.azurecr.io`
  - ACR registry name: `{registry-name}`
  
  If a resource group is configured in the connector, only ACR registries in that
  resource group will be accessible.
  
  If an authentication method other than the Azure service principal is used, Entra ID authentication is used.
  This requires the configured identity to have the `AcrPush` role to be configured.
  If this fails, admin account authentication is tried. For this the admin account must be enabled for the registry.
  See the official Azure[documentation on the admin account](https://docs.microsoft.com/en-us/azure/container-registry/container-registry-authentication#admin-account) for more information. 
  """,
              auth_methods=AzureAuthenticationMethods.values(),
              supports_instances=True,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/docker.png",
              emoji=":whale:",
          ),
      ],
  )
  
  
  class AzureServiceConnector(ServiceConnector):
      """Azure service connector."""
  
      config: AzureBaseConfig
  
      _subscription_id: Optional[str] = None
      _subscription_name: Optional[str] = None
      _tenant_id: Optional[str] = None
      _session_cache: Dict[
          str,
          Tuple[TokenCredential, Optional[datetime.datetime]],
      ] = {}
  
      @classmethod
      def _get_connector_type(cls) -> ServiceConnectorTypeModel:
          """Get the service connector type specification.
  
          Returns:
              The service connector type specification.
          """
          return AZURE_SERVICE_CONNECTOR_TYPE_SPEC
  
      @property
      def subscription(self) -> Tuple[str, str]:
          """Get the Azure subscription ID and name.
  
          Returns:
              The Azure subscription ID and name.
  
          Raises:
              AuthorizationException: If the Azure subscription could not be
                  determined or doesn't match the configured subscription ID.
          """
          if self._subscription_id is None or self._subscription_name is None:
              logger.debug("Getting subscription from Azure...")
              try:
                  credential, _ = self.get_azure_credential(self.auth_method)
                  subscription_client = SubscriptionClient(credential)
                  if self.config.subscription_id is not None:
                      subscription = subscription_client.subscriptions.get(
                          str(self.config.subscription_id)
                      )
                      self._subscription_name = subscription.display_name
                      self._subscription_id = subscription.subscription_id
                  else:
                      subscriptions = list(
                          subscription_client.subscriptions.list()
                      )
                      if len(subscriptions) == 0:
                          raise AuthorizationException(
                              "no Azure subscriptions found for the configured "
                              "credentials."
                          )
                      if len(subscriptions) > 1:
                          raise AuthorizationException(
                              "multiple Azure subscriptions found for the "
                              "configured credentials. Please configure the "
                              "subscription ID explicitly in the connector."
                          )
  
                      subscription = subscriptions[0]
                      self._subscription_id = subscription.subscription_id
                      self._subscription_name = subscription.display_name
              except AzureError as e:
                  raise AuthorizationException(
                      f"failed to fetch the Azure subscription: {e}"
                  ) from e
  
          assert self._subscription_id is not None
          assert self._subscription_name is not None
          return self._subscription_id, self._subscription_name
  
      @property
      def tenant_id(self) -> str:
          """Get the Azure tenant ID.
  
          Returns:
              The Azure tenant ID.
  
          Raises:
              AuthorizationException: If the Azure tenant ID could not be
                  determined or doesn't match the configured tenant ID.
          """
          if self._tenant_id is None:
              logger.debug("Getting tenant ID from Azure...")
              try:
                  credential, _ = self.get_azure_credential(self.auth_method)
                  subscription_client = SubscriptionClient(credential)
                  tenants = subscription_client.tenants.list()
  
                  if self.config.tenant_id is not None:
                      for tenant in tenants:
                          if str(tenant.tenant_id) == str(self.config.tenant_id):
                              self._tenant_id = str(tenant.tenant_id)
                              break
                      else:
                          raise AuthorizationException(
                              "the configured tenant ID is not associated with "
                              "the configured credentials."
                          )
                  else:
                      tenants = list(tenants)
                      if len(tenants) == 0:
                          raise AuthorizationException(
                              "no Azure tenants found for the configured "
                              "credentials."
                          )
                      if len(tenants) > 1:
                          raise AuthorizationException(
                              "multiple Azure tenants found for the "
                              "configured credentials. Please configure the "
                              "tenant ID explicitly in the connector."
                          )
  
                      tenant = tenants[0]
                      self._tenant_id = tenant.tenant_id
              except AzureError as e:
                  raise AuthorizationException(
                      f"failed to fetch the Azure tenant: {e}"
                  ) from e
          assert self._tenant_id is not None
          return self._tenant_id
  
      def get_azure_credential(
          self,
          auth_method: str,
          resource_type: Optional[str] = None,
          resource_id: Optional[str] = None,
      ) -> Tuple[TokenCredential, Optional[datetime.datetime]]:
          """Get an Azure credential for the specified resource.
  
          Args:
              auth_method: The authentication method to use.
              resource_type: The resource type to get a credential for.
              resource_id: The resource ID to get a credential for.
  
          Returns:
              An Azure credential for the specified resource and its expiration
              timestamp, if applicable.
          """
          # We maintain a cache of all sessions to avoid re-authenticating
          # multiple times for the same resource
          key = auth_method
          if key in self._session_cache:
              session, expires_at = self._session_cache[key]
              if expires_at is None:
                  return session, None
  
              # Refresh expired sessions
  
              # check if the token expires in the near future
              if expires_at > utc_now(tz_aware=expires_at) + datetime.timedelta(
                  minutes=AZURE_SESSION_EXPIRATION_BUFFER
              ):
                  return session, expires_at
  
          logger.debug(
              f"Creating Azure credential for auth method '{auth_method}', "
              f"resource type '{resource_type}' and resource ID "
              f"'{resource_id}'..."
          )
          session, expires_at = self._authenticate(
              auth_method, resource_type, resource_id
          )
          self._session_cache[key] = (session, expires_at)
          return session, expires_at
  
      def _authenticate(
          self,
          auth_method: str,
          resource_type: Optional[str] = None,
          resource_id: Optional[str] = None,
      ) -> Tuple[TokenCredential, Optional[datetime.datetime]]:
          """Authenticate to Azure and return a token credential.
  
          Args:
              auth_method: The authentication method to use.
              resource_type: The resource type to authenticate for.
              resource_id: The resource ID to authenticate for.
  
          Returns:
              An Azure token credential and the expiration time of the
              temporary credentials if applicable.
  
          Raises:
              NotImplementedError: If the authentication method is not supported.
              AuthorizationException: If the authentication failed.
          """
          cfg = self.config
          credential: TokenCredential
          if auth_method == AzureAuthenticationMethods.IMPLICIT:
              self._check_implicit_auth_method_allowed()
  
              try:
                  credential = DefaultAzureCredential()
              except AzureError as e:
                  raise AuthorizationException(
                      f"Failed to authenticate to Azure using implicit "
                      f"authentication. Please check your local Azure CLI "
                      f"configuration or managed identity setup: {e}"
                  )
  
              return credential, None
  
          if auth_method == AzureAuthenticationMethods.SERVICE_PRINCIPAL:
              assert isinstance(cfg, AzureServicePrincipalConfig)
              try:
                  credential = ClientSecretCredential(
                      tenant_id=str(cfg.tenant_id),
                      client_id=str(cfg.client_id),
                      client_secret=cfg.client_secret.get_secret_value(),
                  )
              except AzureError as e:
                  raise AuthorizationException(
                      f"Failed to authenticate to Azure using the provided "
                      f"service principal credentials. Please check your Azure "
                      f"configuration: {e}"
                  )
  
              return credential, None
  
          if auth_method == AzureAuthenticationMethods.ACCESS_TOKEN:
              assert isinstance(cfg, AzureAccessTokenConfig)
  
              expires_at = self.expires_at
              assert expires_at is not None
  
              credential = ZenMLAzureTokenCredential(
                  token=cfg.token.get_secret_value(),
                  expires_at=expires_at,
              )
  
              return credential, expires_at
  
          raise NotImplementedError(
              f"Authentication method '{auth_method}' is not supported by "
              "the Azure connector."
          )
  
      def _parse_blob_container_resource_id(self, resource_id: str) -> str:
          """Validate and convert an Azure blob resource ID to a container name.
  
          The resource ID could mean different things:
  
          - Azure blob container URI: `{az|abfs}://{container-name}`
          - Azure blob container name: `{container-name}`
  
          This method extracts the container name from the provided resource ID.
  
          Args:
              resource_id: The resource ID to convert.
  
          Returns:
              The container name.
  
          Raises:
              ValueError: If the provided resource ID is not a valid Azure blob
                  resource ID.
          """
          container_name: Optional[str] = None
          if re.match(
              r"^(az|abfs)://[a-z0-9](?!.*--)[a-z0-9-]{1,61}[a-z0-9](/.*)*$",
              resource_id,
          ):
              # The resource ID is an Azure blob container URI
              container_name = resource_id.split("/")[2]
          elif re.match(
              r"^[a-z0-9](?!.*--)[a-z0-9-]{1,61}[a-z0-9]$",
              resource_id,
          ):
              # The resource ID is the Azure blob container name
              container_name = resource_id
          else:
              raise ValueError(
                  f"Invalid resource ID for an Azure blob storage container: "
                  f"{resource_id}. "
                  "Supported formats are:\n"
                  "Azure blob container URI: {az|abfs}://<container-name>\n"
                  "Azure blob container name: <container-name>"
              )
  
          return container_name
  
      def _parse_acr_resource_id(
          self,
          resource_id: str,
      ) -> str:
          """Validate and convert an ACR resource ID to an ACR registry name.
  
          The resource ID could mean different things:
  
          - ACR registry URI: `[https://]{registry-name}.azurecr.io`
          - ACR registry name: `{registry-name}`
  
          This method extracts the registry name from the provided resource ID.
  
          Args:
              resource_id: The resource ID to convert.
  
          Returns:
              The ACR registry name.
  
          Raises:
              ValueError: If the provided resource ID is not a valid ACR
                  resource ID.
          """
          registry_name: Optional[str] = None
          if re.match(
              r"^(https?://)?[a-zA-Z0-9]+\.azurecr\.io(/.+)*$",
              resource_id,
          ):
              # The resource ID is an ACR registry URI
              registry_name = resource_id.split(".")[0].split("/")[-1]
          elif re.match(
              r"^[a-zA-Z0-9]+$",
              resource_id,
          ):
              # The resource ID is an ACR registry name
              registry_name = resource_id
          else:
              raise ValueError(
                  f"Invalid resource ID for a ACR registry: {resource_id}. "
                  f"Supported formats are:\n"
                  "ACR registry URI: [https://]{registry-name}.azurecr.io\n"
                  "ACR registry name: {registry-name}"
              )
  
          return registry_name
  
      def _parse_aks_resource_id(
          self, resource_id: str
      ) -> Tuple[Optional[str], str]:
          """Validate and convert an AKS resource ID to an AKS cluster name.
  
          The resource ID could mean different things:
  
          - resource group scoped AKS cluster name (canonical): `{resource-group}/{cluster-name}`
          - AKS cluster name: `{cluster-name}`
  
          This method extracts the resource group name and cluster name from the
          provided resource ID.
  
          Args:
              resource_id: The resource ID to convert.
  
          Returns:
              The Azure resource group and AKS cluster name.
  
          Raises:
  