[GitHub] [airflow] elwinarens removed a comment on issue #8258: Implement Azure key vault secrets backend

[GitBox <gi...@apache.org>]

elwinarens removed a comment on issue #8258:
URL: https://github.com/apache/airflow/issues/8258#issuecomment-629216849


   ```
   from typing import Optional
   import re
   
   from azure.identity import DefaultAzureCredential
   from azure.keyvault.secrets import SecretClient
   from cached_property import cached_property
   
   from airflow.secrets import BaseSecretsBackend
   from airflow.utils.log.logging_mixin import LoggingMixin
   
   
   class AzureKeyVaultBackend(BaseSecretsBackend, LoggingMixin):
       """
       Retrieves Airflow Connections or Variables from Azure Key Vault's secrets.
   
       Configurable via ``airflow.cfg`` like so:
   
       .. code-block:: ini
   
           [secrets]
           backend = orchestrator.providers.microsoft.azure.secrets.azure_key_vault.KeyVaultBackend
           backend_kwargs = {"vault_url": "<azure_key_vault_uri>"}
   
       For example, if secrets prefix is ``airflow-connections-smtp-default``, this would be accessible
       if you provide ``{"connections_prefix": "airflow-connections"}`` and request conn_id ``smtp-default``.
       And if variables prefix is ``airflow-variables-hello``, this would be accessible
       if you provide ``{"variables_prefix": "airflow-variables"}`` and request variable key ``hello``.
   
       :param vault_url: The URL of an Azure Key Vault to use.
       :type vault_url: str
       :param connections_prefix: Specifies the prefix of the secret to read to get Connections.
       :type connections_prefix: str
       :param variables_prefix: Specifies the prefix of the secret to read to get Variables.
       :type variables_prefix: str
       :param sep: separator used to concatenate secret_prefix and secret_id. Default: "-"
       :type sep: str
       """
   
       def __init__(
           self,
           vault_url: str = None,
           connections_prefix: str = 'airflow-connections',
           variables_prefix: str = 'airflow-variables',
           sep: str = "-",
           **kwargs
       ):
           super().__init__(**kwargs)
           self.connections_prefix = connections_prefix.rstrip(sep)
           self.variables_prefix = variables_prefix.rstrip(sep)
           self.vault_url = vault_url
           self.sep = sep
           self.kwargs = kwargs
   
       @cached_property
       def client(self):
           """
           Create a Azure Key Vault client
           """
           credential = DefaultAzureCredential()
           client = SecretClient(
               vault_url=self.vault_url,
               credential=credential,
               **self.kwargs
           )
           return client
   
       def get_conn_uri(self, conn_id: str) -> Optional[str]:
           """
           Get an Airflow Connection URI from an Azure Key Vault secret.
   
           :param conn_id: connection id
           :type conn_id: str
           """
           return self._get_secret(self.connections_prefix, conn_id)
   
       def get_variable(self, key: str) -> Optional[str]:
           """
           Get an Airflow Variable from an Azure Key Vault secret.
   
           :param key: Variable Key
           :return: Variable Value
           """
           return self._get_secret(self.variables_prefix, key)
   
       @staticmethod
       def build_path(path_prefix: str, secret_id: str, sep: str = "-") -> str:
           """
           Given a path_prefix and secret_id, build a valid secret bane for the Key Vault Backend.
           Also replaces all non alphanumeric characters in the path with dashes,
           so ``connection_default`` becomes ``connection-default``.
   
           :param path_prefix: Prefix of the path to get secret
           :type path_prefix: str
           :param secret_id: Secret id
           :type secret_id: str
           :param sep: separator used to concatenate path_prefix and secret_id. Default: "-"
           :type sep: str
           """
           path = "{}{}{}".format(path_prefix, sep, secret_id)
           return re.sub('[^0-9a-zA-Z-]', sep, path)
   
       def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
           secret_name = self.build_path(path_prefix, secret_id, self.sep)
           try:
               secret = self.client.get_secret(secret_name)
               return secret.value
           except Exception as ex:
               self.log.debug(
                   "An error occurred when calling the "
                   "get_secret operation: "
                   "Secret %s not found: %s", (secret_name, ex)
               )
               return None
   ```


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org