Azure Key Vault case insensitive support and dash-underscore translation (#607)

Co-authored-by: Oleksa <[email protected]>
Co-authored-by: Hasan Ramezani <[email protected]>

Author: 3 people

docs/index.md

@@ -1889,6 +1889,52 @@ class AzureKeyVaultSettings(BaseSettings):
         )
 ```
 
+### Dash to underscore mapping
+
+The Azure Key Vault source accepts a `dash_to_underscore` option, disabled by default, to support Key Vault kebab-case secret names by mapping them to Python's snake_case field names. When enabled, dashes (`-`) in secret names are mapped to underscores (`_`) in field names during validation.
+
+This mapping applies only to *field names*, not to aliases.
+
+```py
+import os
+
+from azure.identity import DefaultAzureCredential
+from pydantic import Field
+
+from pydantic_settings import (
+    AzureKeyVaultSettingsSource,
+    BaseSettings,
+    PydanticBaseSettingsSource,
+)
+
+
+class AzureKeyVaultSettings(BaseSettings):
+    field_with_underscore: str
+    field_with_alias: str = Field(..., alias='Alias-With-Dashes')
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        az_key_vault_settings = AzureKeyVaultSettingsSource(
+            settings_cls,
+            os.environ['AZURE_KEY_VAULT_URL'],
+            DefaultAzureCredential(),
+            dash_to_underscore=True,
+        )
+        return (az_key_vault_settings,)
+```
+
+This setup will load Azure Key Vault secrets named `field-with-underscore` and `Alias-With-Dashes`, mapping them to the `field_with_underscore` and `field_with_alias` fields, respectively.
+
+!!! tip
+    Alternatively, you can configure an [alias_generator](alias.md#using-alias-generators) to map PascalCase secrets.
+
 ## Google Cloud Secret Manager
 
 Google Cloud Secret Manager allows you to store, manage, and access sensitive information as secrets in Google Cloud Platform. This integration lets you retrieve secrets directly from GCP Secret Manager for use in your Pydantic settings.

pydantic_settings/sources/providers/azure.py

@@ -5,6 +5,8 @@
 from collections.abc import Iterator, Mapping
 from typing import TYPE_CHECKING, Optional
 
+from pydantic.fields import FieldInfo
+
 from .env import EnvSettingsSource
 
 if TYPE_CHECKING:
@@ -42,27 +44,33 @@ class AzureKeyVaultMapping(Mapping[str, Optional[str]]):
     def __init__(
         self,
         secret_client: SecretClient,
+        case_sensitive: bool,
     ) -> None:
         self._loaded_secrets = {}
         self._secret_client = secret_client
-        self._secret_names: list[str] = [
+        self._case_sensitive = case_sensitive
+        self._secret_map: dict[str, str] = self._load_remote()
+
+    def _load_remote(self) -> dict[str, str]:
+        secret_names: Iterator[str] = (
             secret.name for secret in self._secret_client.list_properties_of_secrets() if secret.name and secret.enabled
-        ]
+        )
+        if self._case_sensitive:
+            return {name: name for name in secret_names}
+        return {name.lower(): name for name in secret_names}
 
     def __getitem__(self, key: str) -> str | None:
-        if key not in self._loaded_secrets and key in self._secret_names:
-            try:
-                self._loaded_secrets[key] = self._secret_client.get_secret(key).value
-            except Exception:
-                raise KeyError(key)
-
+        if not self._case_sensitive:
+            key = key.lower()
+        if key not in self._loaded_secrets and key in self._secret_map:
+            self._loaded_secrets[key] = self._secret_client.get_secret(self._secret_map[key]).value
         return self._loaded_secrets[key]
 
     def __len__(self) -> int:
-        return len(self._secret_names)
+        return len(self._secret_map)
 
     def __iter__(self) -> Iterator[str]:
-        return iter(self._secret_names)
+        return iter(self._secret_map.keys())
 
 
 class AzureKeyVaultSettingsSource(EnvSettingsSource):
@@ -74,16 +82,19 @@ def __init__(
         settings_cls: type[BaseSettings],
         url: str,
         credential: TokenCredential,
+        dash_to_underscore: bool = False,
+        case_sensitive: bool | None = None,
         env_prefix: str | None = None,
         env_parse_none_str: str | None = None,
         env_parse_enums: bool | None = None,
     ) -> None:
         import_azure_key_vault()
         self._url = url
         self._credential = credential
+        self._dash_to_underscore = dash_to_underscore
         super().__init__(
             settings_cls,
-            case_sensitive=True,
+            case_sensitive=case_sensitive,
             env_prefix=env_prefix,
             env_nested_delimiter='--',
             env_ignore_empty=False,
@@ -93,7 +104,12 @@ def __init__(
 
     def _load_env_vars(self) -> Mapping[str, Optional[str]]:
         secret_client = SecretClient(vault_url=self._url, credential=self._credential)
-        return AzureKeyVaultMapping(secret_client)
+        return AzureKeyVaultMapping(secret_client, self.case_sensitive)
+
+    def _extract_field_info(self, field: FieldInfo, field_name: str) -> list[tuple[str, str, bool]]:
+        if self._dash_to_underscore:
+            return list((x[0], x[1].replace('_', '-'), x[2]) for x in super()._extract_field_info(field, field_name))
+        return super()._extract_field_info(field, field_name)
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__}(url={self._url!r}, env_nested_delimiter={self.env_nested_delimiter!r})'

tests/test_source_azure_key_vault.py

@@ -160,3 +160,48 @@ def _raise_resource_not_found_when_getting_parent_secret_name(self, secret_name:
             raise ResourceNotFoundError()
 
         return key_vault_secret
+
+    def test_dash_to_underscore_translation(self, mocker: MockerFixture) -> None:
+        """Test that dashes in secret names are mapped to underscores in field names."""
+
+        class AzureKeyVaultSettings(BaseSettings):
+            my_field: str
+            alias_field: str = Field(..., alias='Secret-Alias')
+
+            @classmethod
+            def settings_customise_sources(
+                cls,
+                settings_cls: type[BaseSettings],
+                init_settings: PydanticBaseSettingsSource,
+                env_settings: PydanticBaseSettingsSource,
+                dotenv_settings: PydanticBaseSettingsSource,
+                file_secret_settings: PydanticBaseSettingsSource,
+            ) -> tuple[PydanticBaseSettingsSource, ...]:
+                return (
+                    AzureKeyVaultSettingsSource(
+                        settings_cls,
+                        'https://my-resource.vault.azure.net/',
+                        DefaultAzureCredential(),
+                        dash_to_underscore=True,
+                    ),
+                )
+
+        expected_secrets = [
+            type('', (), {'name': 'my-field', 'enabled': True}),
+            type('', (), {'name': 'Secret-Alias', 'enabled': True}),
+        ]
+        expected_secret_value = 'SecretValue'
+
+        mocker.patch(
+            f'{AzureKeyVaultSettingsSource.__module__}.{SecretClient.list_properties_of_secrets.__qualname__}',
+            return_value=expected_secrets,
+        )
+        mocker.patch(
+            f'{AzureKeyVaultSettingsSource.__module__}.{SecretClient.get_secret.__qualname__}',
+            return_value=KeyVaultSecret(SecretProperties(), expected_secret_value),
+        )
+
+        settings = AzureKeyVaultSettings()
+
+        assert settings.my_field == expected_secret_value
+        assert settings.alias_field == expected_secret_value