Fix error messages in PythonVirtualenvOperator when Azure Key Vault secret backend is configured

providers/microsoft/azure/src/airflow/providers/microsoft/azure/secrets/key_vault.py

@@ -238,6 +238,16 @@ def _get_secret(self, path_prefix: str, secret_id: str, team_name: str | None =
     def _get_secret_value(self, path_prefix: str, secret_id: str) -> str | None:
         """Get an Azure Key Vault secret value for the given prefix and key."""
         name = self.build_path(path_prefix, secret_id, self.sep)
+        # Azure Key Vault secret names must be 1-127 characters, containing only 0-9, a-z, A-Z, and -.
+        # https://learn.microsoft.com/en-us/azure/key-vault/general/about-keys-secrets-certificates#object-identifiers
+        if not re.fullmatch(r"[0-9a-zA-Z-]{1,127}", name):
+            self.log.warning(
+                "Secret name %r is not valid. "
+                "Azure Key Vault secret names must be 1-127 characters long "
+                "and contain only alphanumeric characters and dashes.",
+                name,
+            )
+            return None
         try:
             secret = self.client.get_secret(name=name)
             return secret.value

providers/microsoft/azure/tests/unit/microsoft/azure/secrets/test_key_vault.py

@@ -188,3 +188,13 @@ def test_client_authenticate_with_client_secret_credential(
         backend.client
         assert not mock_defaul_azure_credential.called
         mock_client_secret_credential.assert_called_once()
+
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_get_variable_returns_none_for_invalid_secret_name(self, mock_client):
+        """
+        Test that if the variable key produces an invalid Azure Key Vault secret name
+        (e.g. contains dots), the backend returns None without calling the API.
+        """
+        backend = AzureKeyVaultBackend()
+        assert backend.get_variable("SomeOperator.cache_key") is None
+        mock_client.get_secret.assert_not_called()

providers/standard/docs/changelog.rst

@@ -35,6 +35,13 @@
 Changelog
 ---------
 
+.. warning::
+  The Airflow variable used by ``PythonVirtualenvOperator`` to override the virtualenv cache hash
+  has been renamed from ``PythonVirtualenvOperator.cache_key`` to ``python_virtualenv_operator_cache_key``.
+  The old name contained a dot which is incompatible with secrets backends that restrict allowed
+  characters (e.g. Azure Key Vault). Cached virtual environments will be rebuilt once due to the
+  changed hash input.
+
 1.13.0
 ......
 

providers/standard/docs/operators/python.rst

@@ -257,7 +257,7 @@ different workers, it might happen that virtual environment are created on multi
 of the worker will drop the cache (assuming ``venv_cache_path`` is not on a persistent volume).
 
 In case you have problems during runtime with broken cached virtual environments, you can influence the cache directory hash by setting the Airflow variable
-``PythonVirtualenvOperator.cache_key`` to any text. The content of this variable is uses in the vector to calculate the cache directory key.
+``python_virtualenv_operator_cache_key`` to any text. The content of this variable is used in the vector to calculate the cache directory key.
 
 Note that any modification of a cached virtual environment (like temp files in binary path, post-installing further requirements) might pollute a cached virtual environment and the
 operator is not maintaining or cleaning the cache path.

providers/standard/src/airflow/providers/standard/operators/python.py

@@ -954,7 +954,7 @@ def _calculate_cache_hash(self, exclude_cloudpickle: bool = False) -> tuple[str,
             "requirements_list": self._requirements_list(exclude_cloudpickle=exclude_cloudpickle),
             "pip_install_options": self.pip_install_options,
             "index_urls": self.index_urls,
-            "cache_key": str(Variable.get("PythonVirtualenvOperator.cache_key", "")),
+            "cache_key": str(Variable.get("python_virtualenv_operator_cache_key", "")),
             "python_version": self.python_version,
             "system_site_packages": self.system_site_packages,
         }