Skip to content

Base

Base for Parameter providers

Usage Documentation

Parameters

CLASS DESCRIPTION
BaseProvider

Abstract Base Class for Parameter providers

FUNCTION DESCRIPTION
clear_caches

Clear cached parameter values from all providers

get_transform_method

Determine the transform method

transform_value

Transform a value using one of the available options.

BaseProvider

BaseProvider(*, client=None, resource=None)

Bases: ABC

Abstract Base Class for Parameter providers

METHOD DESCRIPTION
get

Retrieve a parameter value or return the cached value

get_multiple

Retrieve multiple parameters based on a path prefix

set

Set parameter value from the underlying parameter store

Source code in aws_lambda_powertools/utilities/parameters/base.py
43
44
45
46
47
48
49
50
51
52
def __init__(self, *, client=None, resource=None):
    """
    Initialize the base provider
    """
    if client is not None:
        user_agent.register_feature_to_client(client=client, feature="parameters")
    if resource is not None:
        user_agent.register_feature_to_resource(resource=resource, feature="parameters")

    self.store: dict[tuple, ExpirableValue] = {}

get

get(name: str, max_age: int | None = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options) -> str | bytes | dict | None

Retrieve a parameter value or return the cached value

PARAMETER DESCRIPTION
name

Parameter name

TYPE: str

max_age

Maximum age of the cached value

TYPE: int | None DEFAULT: None

transform

Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values.

TYPE: TransformOptions DEFAULT: None

force_fetch

Force update even before a cached item has expired, defaults to False

TYPE: bool DEFAULT: False

sdk_options

Arguments that will be passed directly to the underlying API call

DEFAULT: {}

RAISES DESCRIPTION
GetParameterError

When the parameter provider fails to retrieve a parameter value for a given name.

TransformParameterError

When the parameter provider fails to transform a parameter value.

Source code in aws_lambda_powertools/utilities/parameters/base.py
 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
def get(
    self,
    name: str,
    max_age: int | None = None,
    transform: TransformOptions = None,
    force_fetch: bool = False,
    **sdk_options,
) -> str | bytes | dict | None:
    """
    Retrieve a parameter value or return the cached value

    Parameters
    ----------
    name: str
        Parameter name
    max_age: int
        Maximum age of the cached value
    transform: str
        Optional transformation of the parameter value. Supported values
        are "json" for JSON strings and "binary" for base 64 encoded
        values.
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Arguments that will be passed directly to the underlying API call

    Raises
    ------
    GetParameterError
        When the parameter provider fails to retrieve a parameter value for
        a given name.
    TransformParameterError
        When the parameter provider fails to transform a parameter value.
    """

    # If there are multiple calls to the same parameter but in a different
    # transform, they will be stored multiple times. This allows us to
    # optimize by transforming the data only once per retrieval, thus there
    # is no need to transform cached values multiple times. However, this
    # means that we need to make multiple calls to the underlying parameter
    # store if we need to return it in different transforms. Since the number
    # of supported transform is small and the probability that a given
    # parameter will always be used in a specific transform, this should be
    # an acceptable tradeoff.
    value: str | bytes | dict | None = None
    key = self._build_cache_key(name=name, transform=transform, sdk_options=sdk_options)

    # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)

    if not force_fetch and self.has_not_expired_in_cache(key):
        return self.fetch_from_cache(key)

    try:
        value = self._get(name, **sdk_options)
    # Encapsulate all errors into a generic GetParameterError
    except Exception as exc:
        raise GetParameterError(str(exc))

    if transform:
        value = transform_value(key=name, value=value, transform=transform, raise_on_transform_error=True)

    # NOTE: don't cache None, as they might've been failed transforms and may be corrected
    if value is not None:
        self.add_to_cache(key=key, value=value, max_age=max_age)

    return value

get_multiple

get_multiple(path: str, max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, force_fetch: bool = False, **sdk_options) -> dict[str, str] | dict[str, bytes] | dict[str, dict]

Retrieve multiple parameters based on a path prefix

PARAMETER DESCRIPTION
path

Parameter path used to retrieve multiple parameters

TYPE: str

max_age

Maximum age of the cached value

TYPE: int | None DEFAULT: None

transform

Optional transformation of the parameter value. Supported values are "json" for JSON strings, "binary" for base 64 encoded values or "auto" which looks at the attribute key to determine the type.

TYPE: TransformOptions DEFAULT: None

raise_on_transform_error

Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed

TYPE: bool DEFAULT: False

force_fetch

Force update even before a cached item has expired, defaults to False

TYPE: bool DEFAULT: False

sdk_options

Arguments that will be passed directly to the underlying API call

DEFAULT: {}

RAISES DESCRIPTION
GetParameterError

When the parameter provider fails to retrieve parameter values for a given path.

TransformParameterError

When the parameter provider fails to transform a parameter value.

Source code in aws_lambda_powertools/utilities/parameters/base.py
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
193
194
195
196
197
def get_multiple(
    self,
    path: str,
    max_age: int | None = None,
    transform: TransformOptions = None,
    raise_on_transform_error: bool = False,
    force_fetch: bool = False,
    **sdk_options,
) -> dict[str, str] | dict[str, bytes] | dict[str, dict]:
    """
    Retrieve multiple parameters based on a path prefix

    Parameters
    ----------
    path: str
        Parameter path used to retrieve multiple parameters
    max_age: int, optional
        Maximum age of the cached value
    transform: str, optional
        Optional transformation of the parameter value. Supported values
        are "json" for JSON strings, "binary" for base 64 encoded
        values or "auto" which looks at the attribute key to determine the type.
    raise_on_transform_error: bool, optional
        Raises an exception if any transform fails, otherwise this will
        return a None value for each transform that failed
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Arguments that will be passed directly to the underlying API call

    Raises
    ------
    GetParameterError
        When the parameter provider fails to retrieve parameter values for
        a given path.
    TransformParameterError
        When the parameter provider fails to transform a parameter value.
    """
    key = self._build_cache_key(name=path, transform=transform, is_nested=True, sdk_options=sdk_options)

    # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)

    if not force_fetch and self.has_not_expired_in_cache(key):
        return self.fetch_from_cache(key)

    try:
        values = self._get_multiple(path, **sdk_options)
    # Encapsulate all errors into a generic GetParameterError
    except Exception as exc:
        raise GetParameterError(str(exc))

    if transform:
        values.update(
            transform_value(value=values, transform=transform, raise_on_transform_error=raise_on_transform_error),
        )

    self.add_to_cache(key=key, value=values, max_age=max_age)

    return values

set

set(name: str, value: Any, *, overwrite: bool = False, **kwargs)

Set parameter value from the underlying parameter store

Source code in aws_lambda_powertools/utilities/parameters/base.py
132
133
134
135
136
def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs):
    """
    Set parameter value from the underlying parameter store
    """
    raise NotImplementedError()

clear_caches

clear_caches()

Clear cached parameter values from all providers

Source code in aws_lambda_powertools/utilities/parameters/base.py
392
393
394
def clear_caches():
    """Clear cached parameter values from all providers"""
    DEFAULT_PROVIDERS.clear()

get_transform_method

get_transform_method(value: str, transform: TransformOptions = None) -> Callable[..., Any]

Determine the transform method

Examples:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
>>> get_transform_method("key","any_other_value")
'any_other_value'
>>> get_transform_method("key.json","auto")
'json'
>>> get_transform_method("key.binary","auto")
'binary'
>>> get_transform_method("key","auto")
None
>>> get_transform_method("key",None)
None
PARAMETER DESCRIPTION
value

Only used when the transform is "auto".

TYPE: str

transform

Original transform method, only "auto" will try to detect the transform method by the key

TYPE: TransformOptions DEFAULT: None

RETURNS DESCRIPTION
Callable

Transform function could be json.loads, base64.b64decode, or a lambda that echo the str value

TYPE: Callable[..., Any]

Source code in aws_lambda_powertools/utilities/parameters/base.py
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def get_transform_method(value: str, transform: TransformOptions = None) -> Callable[..., Any]:
    """
    Determine the transform method

    Examples
    -------
        >>> get_transform_method("key","any_other_value")
        'any_other_value'
        >>> get_transform_method("key.json","auto")
        'json'
        >>> get_transform_method("key.binary","auto")
        'binary'
        >>> get_transform_method("key","auto")
        None
        >>> get_transform_method("key",None)
        None

    Parameters
    ---------
    value: str
        Only used when the transform is "auto".
    transform: str, optional
        Original transform method, only "auto" will try to detect the transform method by the key

    Returns
    ------
    Callable:
        Transform function could be json.loads, base64.b64decode, or a lambda that echo the str value
    """
    transform_method = TRANSFORM_METHOD_MAPPING.get(transform)

    if transform == "auto":
        key_suffix = value.rsplit(".", maxsplit=1)[-1]
        transform_method = TRANSFORM_METHOD_MAPPING.get(key_suffix, TRANSFORM_METHOD_MAPPING[None])

    return cast(Callable, transform_method)  # https://github.com/python/mypy/issues/10740

transform_value

transform_value(value: dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = False, key: str = '') -> dict[str, Any]
transform_value(value: str | bytes | dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = False, key: str = '') -> str | bytes | dict[str, Any] | None
transform_value(value: str | bytes | dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = True, key: str = '') -> str | bytes | dict[str, Any] | None

Transform a value using one of the available options.

PARAMETER DESCRIPTION
value

Parameter value to transform

TYPE: str | bytes | dict[str, Any]

transform

Type of transform, supported values are "json", "binary", and "auto" based on suffix (.json, .binary)

TYPE: TransformOptions

key

Parameter key when transform is auto to infer its transform method

TYPE: str DEFAULT: ''

raise_on_transform_error

Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed

TYPE: bool DEFAULT: True

RAISES DESCRIPTION
TransformParameterError:

When the parameter value could not be transformed

Source code in aws_lambda_powertools/utilities/parameters/base.py
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
def transform_value(
    value: str | bytes | dict[str, Any],
    transform: TransformOptions,
    raise_on_transform_error: bool = True,
    key: str = "",
) -> str | bytes | dict[str, Any] | None:
    """
    Transform a value using one of the available options.

    Parameters
    ---------
    value: str
        Parameter value to transform
    transform: str
        Type of transform, supported values are "json", "binary", and "auto" based on suffix (.json, .binary)
    key: str
        Parameter key when transform is auto to infer its transform method
    raise_on_transform_error: bool, optional
        Raises an exception if any transform fails, otherwise this will
        return a None value for each transform that failed

    Raises
    ------
    TransformParameterError:
        When the parameter value could not be transformed
    """
    # Maintenance: For v3, we should consider returning the original value for soft transform failures.

    err_msg = "Unable to transform value using '{transform}' transform: {exc}"

    if isinstance(value, dict):
        # NOTE: We must handle partial failures when receiving multiple values
        # where one of the keys might fail during transform, e.g. `{"a": "valid", "b": "{"}`
        # expected: `{"a": "valid", "b": None}`

        transformed_values: dict[str, Any] = {}
        for dict_key, dict_value in value.items():
            transform_method = get_transform_method(value=dict_key, transform=transform)
            try:
                transformed_values[dict_key] = transform_method(dict_value)
            except Exception as exc:
                if raise_on_transform_error:
                    raise TransformParameterError(err_msg.format(transform=transform, exc=exc)) from exc
                transformed_values[dict_key] = None
        return transformed_values

    # Auto transforms inspect the parameter name; explicit transforms ignore it.
    transform_method = get_transform_method(value=key, transform=transform)

    try:
        # Identity transforms must preserve raw bytes. Decode failures belong to
        # the same error policy as JSON/base64 failures.
        if isinstance(value, bytes) and transform_method is not TRANSFORM_METHOD_MAPPING[None]:
            value = value.decode("utf-8")
        return transform_method(value)
    except Exception as exc:
        if raise_on_transform_error:
            raise TransformParameterError(err_msg.format(transform=transform, exc=exc)) from exc
        return None