Skip to content

AWS Secrets Manager

Bases: PluginResolver, AWSSecretManagerMixin

Resolver for the AWS Secrets Manager

Methods:

Name Description
__init__

Initializes the resolver with a boto3 Session. If no Session is provided, it'll be infered from the default credentials, if configured.
__call__

Resolves the secret by its name and returns the decoded secret data.
Example

Example 1: Retrieve a secret as a string 

>>> resolver = AWSSecretsManagerResolver(session=boto3_session, infer_json=True)
>>> secret_value = resolver('my_secret')
>>> print(secret_value) # {"a": 1}

Example 2: Retrieve a secret and parse it as JSON 

>>> resolver = AWSSecretsManagerResolver(session=boto3_session, infer_json=False)
>>> secret_value = resolver('my_secret')
>>> print(secret_value) # '{"a": 1}'

Source code in omegaconf_cloud_resolvers/resolvers/aws/secretsmanager.py 
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class AWSSecretsManagerResolver(PluginResolver, AWSSecretManagerMixin):
    """
    Resolver for the AWS Secrets Manager

    Methods:
        __init__(session=None, infer_json=False, return_binary=False, *args, **kwargs):
            Initializes the resolver with a boto3 Session.
            If no Session is provided, it'll be infered from the default credentials,
            if configured.

        __call__(name):
            Resolves the secret by its name and returns the decoded secret data.

    Example:
        Example 1: Retrieve a secret as a string
        ```python
        >>> resolver = AWSSecretsManagerResolver(session=boto3_session, infer_json=True)
        >>> secret_value = resolver('my_secret')
        >>> print(secret_value) # {"a": 1}
        ```

        Example 2: Retrieve a secret and parse it as JSON
        ```python
        >>> resolver = AWSSecretsManagerResolver(session=boto3_session, infer_json=False)
        >>> secret_value = resolver('my_secret')
        >>> print(secret_value) # '{"a": 1}'
        ```


    """

    def __init__(
        self,
        session=None,
        infer_json: bool = False,
        return_binary: bool = False,
        *args,
        **kwargs,
    ):
        """
        Initializes the AWSSecretsManagerResolver.

        Args:
            session (boto3.Session): boto3.Session to use for AWS interactions. If none provided, tries to use the default configuration.
            infer_json: If True, tries to parse the secret as JSON during the __call__.
            return_binary: If True, tries to return the binary value from the key `SecretBinary` instead of `SecretString`.
        """
        super().__init__(session, *args, **kwargs)
        self._infer_json = infer_json
        self._return_binary = return_binary
        if self._return_binary and self._infer_json:
            logger.warning("infer_json is only tried for string secrets")

    def __call__(self, name: str) -> JsonType:
        """
        Retrieves a secret from AWS Secrets Manager.
        The default behaviour prioritizes 'SecretString' over 'SecretBinary'

        Args:
            name: The name of the secret to retrieve.

        Returns:
            The secret value. If infer_json is True and the secret is a valid JSON string,
                         it returns a dictionary. Otherwise, it returns the secret as a string.

        Raises:
            KeyError: If the secret does not contain either of "SecretString", "SecretBinary" key.
        """

        secret = self.client.get_secret_value(SecretId=name)

        try:
            if self._return_binary:
                return secret["SecretBinary"]
            else:
                secret = secret["SecretString"]
                if self._infer_json:
                    return try_cast_to_dict(secret)
                else:
                    return secret
        except KeyError as e:
            logger.exception("The found secret does not contain '%s'", e.args[0])
            raise

__call__(name)

Retrieves a secret from AWS Secrets Manager. The default behaviour prioritizes 'SecretString' over 'SecretBinary'

Parameters:

Name Type Description Default
name str

The name of the secret to retrieve.

 required

Returns:

Type Description
JsonType

The secret value. If infer_json is True and the secret is a valid JSON string, it returns a dictionary. Otherwise, it returns the secret as a string.

Raises:

Type Description
KeyError

If the secret does not contain either of "SecretString", "SecretBinary" key.
Source code in omegaconf_cloud_resolvers/resolvers/aws/secretsmanager.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __call__(self, name: str) -> JsonType:
    """
    Retrieves a secret from AWS Secrets Manager.
    The default behaviour prioritizes 'SecretString' over 'SecretBinary'

    Args:
        name: The name of the secret to retrieve.

    Returns:
        The secret value. If infer_json is True and the secret is a valid JSON string,
                     it returns a dictionary. Otherwise, it returns the secret as a string.

    Raises:
        KeyError: If the secret does not contain either of "SecretString", "SecretBinary" key.
    """

    secret = self.client.get_secret_value(SecretId=name)

    try:
        if self._return_binary:
            return secret["SecretBinary"]
        else:
            secret = secret["SecretString"]
            if self._infer_json:
                return try_cast_to_dict(secret)
            else:
                return secret
    except KeyError as e:
        logger.exception("The found secret does not contain '%s'", e.args[0])
        raise

__init__(session=None, infer_json=False, return_binary=False, *args, **kwargs)

Initializes the AWSSecretsManagerResolver.

Parameters:

Name Type Description Default
session Session

boto3.Session to use for AWS interactions. If none provided, tries to use the default configuration.

 None
infer_json bool

If True, tries to parse the secret as JSON during the call.

 False
return_binary bool

If True, tries to return the binary value from the key SecretBinary instead of SecretString.

 False
Source code in omegaconf_cloud_resolvers/resolvers/aws/secretsmanager.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    session=None,
    infer_json: bool = False,
    return_binary: bool = False,
    *args,
    **kwargs,
):
    """
    Initializes the AWSSecretsManagerResolver.

    Args:
        session (boto3.Session): boto3.Session to use for AWS interactions. If none provided, tries to use the default configuration.
        infer_json: If True, tries to parse the secret as JSON during the __call__.
        return_binary: If True, tries to return the binary value from the key `SecretBinary` instead of `SecretString`.
    """
    super().__init__(session, *args, **kwargs)
    self._infer_json = infer_json
    self._return_binary = return_binary
    if self._return_binary and self._infer_json:
        logger.warning("infer_json is only tried for string secrets")