Skip to content

AWS Parameter Store

Bases: PluginResolver, AWSParameterStoreMixin

Resolver for AWS Systems Manager Parameter Store

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 providing a custom session 

>>> resolver = AWSParameterStoreResolver(session=boto3_session)
>>> parameter_value = resolver('my/parameter/name') # Contains 1 value
>>> print(parameter_value)
Example 2: Retrieve a StringList and try to cast the types 
>>> resolver = AWSParameterStoreResolver(infer_types=True)
>>> parameter_value = resolver('my/parameter/list/name') # Contains 1 value
>>> print(parameter_value)
Example 3: Retrieve a family of parameters without recursion (i.e. 1 level) 
>>> resolver = AWSParameterStoreResolver()
>>> parameter_value = resolver('my/parameter/family/*')
>>> print(parameter_value)
Example 4: Retrieve a family of parameters with recursion 
>>> resolver = AWSParameterStoreResolver()
>>> parameter_value = resolver('my/parameter/family/**')
>>> print(parameter_value)

Source code in omegaconf_cloud_resolvers/resolvers/aws/parameterstore.py 
 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
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class AWSParameterStoreResolver(PluginResolver, AWSParameterStoreMixin):
    """
    Resolver for AWS Systems Manager Parameter Store

    Methods:
        __init__(session=None, decrypt=True, infer_types=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 providing a custom session
        ```python
        >>> resolver = AWSParameterStoreResolver(session=boto3_session)
        >>> parameter_value = resolver('my/parameter/name') # Contains 1 value
        >>> print(parameter_value)
        ```
        Example 2: Retrieve a StringList and try to cast the types
        ```python
        >>> resolver = AWSParameterStoreResolver(infer_types=True)
        >>> parameter_value = resolver('my/parameter/list/name') # Contains 1 value
        >>> print(parameter_value)
        ```
        Example 3: Retrieve a family of parameters without recursion (i.e. 1 level)
        ```python
        >>> resolver = AWSParameterStoreResolver()
        >>> parameter_value = resolver('my/parameter/family/*')
        >>> print(parameter_value)
        ```
        Example 4: Retrieve a family of parameters with recursion
        ```python
        >>> resolver = AWSParameterStoreResolver()
        >>> parameter_value = resolver('my/parameter/family/**')
        >>> print(parameter_value)
        ```
    """

    def __init__(
        self,
        session: Session = None,
        decrypt: bool = True,
        infer_types: bool = False,
        *args,
        **kwargs,
    ):
        """
        Initializes the AWSParameterStoreResolver.

        Args:
            session: An optional boto3 session to use for AWS interactions.
            decrypt: If True, attempts to decrypt SecureString parameters.
            infer_types: If True, attempts to infer the type of the parameter value, it might be
                more useful when the parameter is of type StringList if mixed types are found.
        """
        super().__init__(session, *args, **kwargs)
        self._decrypt = decrypt
        self._infer_types = infer_types

    def __call__(self, name: str) -> JsonType:
        """
        Retrieves a parameter from AWS Systems Manager Parameter Store.

        Args:
            name: The name of the parameter to retrieve.
                Retrieve a single parameter,

        Returns:
            The parameter value. If the parameter is a StringList and _decrypt is False,
                                         it returns a list of strings or numbers (if _infer_types is True).
                                         Otherwise, it returns the parameter value as a string, int, or float
                                         depending on the type inference.
        """
        if name.endswith("/*") or name.endswith("/**"):
            if name.endswith(
                "/**"
            ):  # If '**' access recursively, if only '*' just to the first level.
                recursive = True
                name = name.removesuffix("/**")
            else:
                recursive = False
                name = name.removesuffix("/*")
            return self._get_path_parameters(name, recursive)
        else:
            return self._get_single_parameter(name)

    def _get_single_parameter(self, name: str) -> JsonType:
        value = self.client.get_parameter(Name=name, WithDecryption=self._decrypt)
        ptype = value["Parameter"]["Type"]
        pvalue = value["Parameter"]["Value"]
        return self._parse_value(pvalue, ptype, self._decrypt, self._infer_types)

    def _get_path_parameters(self, name: str, recursive: bool = False) -> JsonType:
        rparameters = []
        # Use the get_parameters_by_path method to retrieve parameters
        response = self.client.get_parameters_by_path(
            Path=name, Recursive=recursive, WithDecryption=self._decrypt
        )
        rparameters.extend(response["Parameters"])

        # Check if there are more parameters to retrieve
        while "NextToken" in response:
            response = self._decrypt.get_parameters_by_path(
                Path=name,
                Recursive=recursive,
                WithDecryption=self._decrypt,
                NextToken=response["NextToken"],
            )
            rparameters.extend(response["Parameters"])

        parameters = {}
        for parameter in rparameters:
            pname, ptype, pvalue = (
                parameter["Name"],
                parameter["Type"],
                parameter["Value"],
            )
            pname = pname.replace(name, "").lstrip("/")
            parameters[pname] = self._parse_value(
                value=pvalue,
                type_=ptype,
                decrypt=self._decrypt,
                infer_types=self._infer_types,
            )
        return parameters

    @staticmethod
    def _parse_value(
        value: str,
        type_: AWS_PARAMETER_TYPE,
        decrypt: bool = False,
        infer_types: bool = False,
    ) -> JsonType:
        """
        Parses the parameter value based on its type and the provided flags.

        Args:
            value (str): The parameter value to parse.
            type_ (Literal["String", "StringList", "SecureString"]): The type of the parameter.
            decrypt (bool): A flag indicating whether the parameter value should be decrypted, for "String", "StringList",
              it does make any difference. For "SecureString", if False, it returns the encrypted string.
            infer_types (bool): A flag indicating whether to attempt to infer and cast the type of the parameter value.
              Note: it does not strip strings, so "a,b, c" will return ["a", "b", " c"]

        Returns:
            The parsed parameter value. If the parameter is a StringList and decrypt is False,
            it returns a list of strings or numbers (if infer_types is True).
            Otherwise, it returns the parameter value as a string, int, or float
            depending on the type inference.
        """

        if type_ == "StringList":
            values = value.split(",")
            if infer_types:
                return [try_infer_and_cast(v) for v in values]
            else:
                return values

        if type_ == "SecureString" and not decrypt:
            if infer_types:
                logger.warning(
                    "infer_types is ignored when parameter type is 'SecureString' and decrypt is False"
                )
            return value  # returns the encrypted value

        if infer_types:
            return try_infer_and_cast(value)
        else:
            return value

__call__(name)

Retrieves a parameter from AWS Systems Manager Parameter Store.

Parameters:

Name Type Description Default
name str

The name of the parameter to retrieve. Retrieve a single parameter,

 required

Returns:

Type Description
JsonType

The parameter value. If the parameter is a StringList and _decrypt is False, it returns a list of strings or numbers (if _infer_types is True). Otherwise, it returns the parameter value as a string, int, or float depending on the type inference.
Source code in omegaconf_cloud_resolvers/resolvers/aws/parameterstore.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __call__(self, name: str) -> JsonType:
    """
    Retrieves a parameter from AWS Systems Manager Parameter Store.

    Args:
        name: The name of the parameter to retrieve.
            Retrieve a single parameter,

    Returns:
        The parameter value. If the parameter is a StringList and _decrypt is False,
                                     it returns a list of strings or numbers (if _infer_types is True).
                                     Otherwise, it returns the parameter value as a string, int, or float
                                     depending on the type inference.
    """
    if name.endswith("/*") or name.endswith("/**"):
        if name.endswith(
            "/**"
        ):  # If '**' access recursively, if only '*' just to the first level.
            recursive = True
            name = name.removesuffix("/**")
        else:
            recursive = False
            name = name.removesuffix("/*")
        return self._get_path_parameters(name, recursive)
    else:
        return self._get_single_parameter(name)

__init__(session=None, decrypt=True, infer_types=False, *args, **kwargs)

Initializes the AWSParameterStoreResolver.

Parameters:

Name Type Description Default
session Session

An optional boto3 session to use for AWS interactions.

 None
decrypt bool

If True, attempts to decrypt SecureString parameters.

 True
infer_types bool

If True, attempts to infer the type of the parameter value, it might be more useful when the parameter is of type StringList if mixed types are found.

 False
Source code in omegaconf_cloud_resolvers/resolvers/aws/parameterstore.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __init__(
    self,
    session: Session = None,
    decrypt: bool = True,
    infer_types: bool = False,
    *args,
    **kwargs,
):
    """
    Initializes the AWSParameterStoreResolver.

    Args:
        session: An optional boto3 session to use for AWS interactions.
        decrypt: If True, attempts to decrypt SecureString parameters.
        infer_types: If True, attempts to infer the type of the parameter value, it might be
            more useful when the parameter is of type StringList if mixed types are found.
    """
    super().__init__(session, *args, **kwargs)
    self._decrypt = decrypt
    self._infer_types = infer_types