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

resolve_parameters_max_age

Resolve the caching max age against the POWERTOOLS_PARAMETERS_MAX_AGE environment variable.

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
44
45
46
47
48
49
50
51
52
53
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
 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
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_parameters_max_age(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
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
198
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_parameters_max_age(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
133
134
135
136
137
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
428
429
430
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
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
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

resolve_parameters_max_age

resolve_parameters_max_age(choice: int | None) -> int

Resolve the caching max age against the POWERTOOLS_PARAMETERS_MAX_AGE environment variable.

An explicit choice always wins. Otherwise the environment variable is used, and an empty or non-integer value is ignored with a :class:PowertoolsUserWarning rather than crashing every lookup: a configuration typo must not break parameter and secret reads (ADR 0006: warn now, harden in a future major).

PARAMETER DESCRIPTION
choice

Explicit max age passed by the caller, taking precedence over the environment.

TYPE: int | None

RETURNS DESCRIPTION
int

The resolved max age in seconds.

Source code in aws_lambda_powertools/utilities/parameters/base.py
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
def resolve_parameters_max_age(choice: int | None) -> int:
    """
    Resolve the caching max age against the ``POWERTOOLS_PARAMETERS_MAX_AGE`` environment variable.

    An explicit ``choice`` always wins. Otherwise the environment variable is used, and an
    empty or non-integer value is ignored with a :class:`PowertoolsUserWarning` rather than
    crashing every lookup: a configuration typo must not break parameter and secret reads
    (ADR 0006: warn now, harden in a future major).

    Parameters
    ----------
    choice : int | None
        Explicit max age passed by the caller, taking precedence over the environment.

    Returns
    -------
    int
        The resolved max age in seconds.
    """
    if choice is not None:
        return choice

    env = os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS)
    try:
        return int(env)
    except ValueError:
        warnings.warn(
            f"Ignoring invalid {constants.PARAMETERS_MAX_AGE_ENV} value {env!r}; "
            f"falling back to the default of {DEFAULT_MAX_AGE_SECS} seconds.",
            category=PowertoolsUserWarning,
            stacklevel=3,
        )
        return int(DEFAULT_MAX_AGE_SECS)

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
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
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