Skip to content

Awssecretsmanager

settus.settingssources.AWSSecretsManager ¤

Bases: PydanticBaseEnvSettingsSource

AWS Secrets settings source class that loads variables from an AWS secrets manager resource.

Functions¤

get_field_value ¤

get_field_value(field, field_name)

Get field value from AWS Secrets Manager

PARAMETER DESCRIPTION
field

Field

TYPE: FieldInfo

field_name

Field name

TYPE: str

RETURNS DESCRIPTION
(field_value, field_key, is_complex)

Output used in __call__ method
Source code in settus/settingssources/awssecretsmanager.py 
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
79
80
81
82
def get_field_value(
    self, field: FieldInfo, field_name: str
) -> Tuple[Any, str, bool]:
    """
    Get field value from AWS Secrets Manager

    Parameters
    ----------
    field:
        Field
    field_name
        Field name

    Returns
    -------
    field_value, field_key, is_complex
        Output used in `__call__` method
    """
    secret_name = None

    # Get keyvault from field
    if field.json_schema_extra is not None:
        secret_name = field.json_schema_extra.get("aws_secret_name")

    # Get keyvault from config
    if secret_name is None:
        secret_name = self.config.get("aws_secret_name")

    if secret_name is None:
        return None, field_name, False

    # Default credentials
    # https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html
    # The most common approach here is to set the following environment variables:
    #  - AWS_ACCESS_KEY_ID
    #  - AWS_SECRET_ACCESS_KEY
    #  - AWS_REGION
    import boto3
    from botocore.exceptions import ClientError

    # Session
    session = boto3.session.Session()

    # Client
    client = session.client(
        service_name="secretsmanager", region_name=os.getenv("AWS_REGION")
    )

    env_val: Union[str, None] = None
    for field_key, env_name, value_is_complex in self._extract_field_info(
        field, field_name
    ):
        try:
            var = client.get_secret_value(SecretId=secret_name)["SecretString"]
            try:
                var = json.loads(var)
                try:
                    env_val = var[env_name]
                except KeyError as e:
                    pass
            except TypeError as e:
                raise TypeError("Secret variable should by type key/value pair")
        except ClientError as e:
            pass
        if env_val is not None:
            break

    return env_val, field_key, value_is_complex