Bases: PydanticBaseEnvSettingsSource
Azure Key Vault settings source class that loads variables from an azure
secrets manager resource.
Functions
get_field_value(field, field_name)
Get field value from keyvault
| PARAMETER |
DESCRIPTION |
field |
TYPE:
FieldInfo
|
field_name |
TYPE:
str
|
| RETURNS |
DESCRIPTION |
(field_value, field_key, is_complex)
|
Output used in __call__ method
|
Source code in settus/settingssources/azurekeyvault.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78 | def get_field_value(
self, field: FieldInfo, field_name: str
) -> Tuple[Any, str, bool]:
"""
Get field value from keyvault
Parameters
----------
field:
Field
field_name
Field name
Returns
-------
field_value, field_key, is_complex
Output used in `__call__` method
"""
keyvault_url = None
keyvault_credentials = None
# Get keyvault from field
if field.json_schema_extra is not None:
keyvault_url = field.json_schema_extra.get("keyvault_url")
keyvault_credentials = field.json_schema_extra.get("keyvault_credentials")
# Get keyvault from config
if keyvault_url is None:
keyvault_url = self.config.get("keyvault_url")
if keyvault_url is None:
return None, field_name, False
if keyvault_credentials is None:
keyvault_credentials = self.config.get("keyvault_credentials")
# Default credentials
# https://learn.microsoft.com/en-us/azure/developer/python/sdk/authentication-overview#sequence-of-authentication-methods-when-you-use-defaultazurecredential
# The most common approach here is to set the following environment variables:
# - AZURE_TENANT_ID
# - AZURE_CLIENT_ID
# - AZURE_CLIENT_SECRET
from azure.core.exceptions import ResourceNotFoundError
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
if keyvault_credentials is None:
keyvault_credentials = DefaultAzureCredential()
# Keyvault client
client = SecretClient(vault_url=keyvault_url, credential=keyvault_credentials)
env_val: Union[str, None] = None
for field_key, env_name, value_is_complex in self._extract_field_info(
field, field_name
):
if "_" in env_name:
continue
try:
env_val = client.get_secret(env_name).value
except (ResourceNotFoundError, HttpResponseError):
env_val = None
if env_val is not None:
break
return env_val, field_key, value_is_complex
|