Module aws_lambda_powertools.utilities.parameters.base

Base for Parameter providers

Functions

def clear_caches()

Clear cached parameter values from all providers

def get_transform_method(value: str, transform: TransformOptions = None)

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

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

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

Classes

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

Abstract Base Class for Parameter providers

Initialize the base provider

Expand source code
class BaseProvider(ABC):
    """
    Abstract Base Class for Parameter providers
    """

    store: dict[tuple, ExpirableValue]

    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] = {}

    def has_not_expired_in_cache(self, key: tuple) -> bool:
        return key in self.store and self.store[key].ttl >= datetime.now()

    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)

        # 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

    @abstractmethod
    def _get(self, name: str, **sdk_options) -> str | bytes | dict[str, Any]:
        """
        Retrieve parameter value from the underlying parameter store
        """
        raise NotImplementedError()

    def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs):
        """
        Set parameter value from the underlying parameter store
        """
        raise NotImplementedError()

    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)

        # 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(values, transform, raise_on_transform_error))

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

        return values

    @abstractmethod
    def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]:
        """
        Retrieve multiple parameter values from the underlying parameter store
        """
        raise NotImplementedError()

    def clear_cache(self):
        self.store.clear()

    def fetch_from_cache(self, key: tuple):
        return self.store[key].value if key in self.store else {}

    def add_to_cache(self, key: tuple, value: Any, max_age: int):
        if max_age <= 0:
            return

        self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))

    def _build_cache_key(
        self,
        name: str,
        transform: TransformOptions = None,
        is_nested: bool = False,
    ):
        """Creates cache key for parameters

        Parameters
        ----------
        name : str
            Name of parameter, secret or config
        transform : TransformOptions, optional
            Transform method used, by default None
        is_nested : bool, optional
            Whether it's a single parameter or multiple nested parameters, by default False

        Returns
        -------
        tuple[str, TransformOptions, bool]
            Cache key
        """
        return (name, transform, is_nested)

Ancestors

  • abc.ABC

Subclasses

Class variables

var store : dict[tuple, ExpirableValue]

Methods

def add_to_cache(self, key: tuple, value: Any, max_age: int)
def clear_cache(self)
def fetch_from_cache(self, key: tuple)
def get(self, name: str, max_age: int | None = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options)

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.
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)

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.
def has_not_expired_in_cache(self, key: tuple) ‑> bool
def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs)

Set parameter value from the underlying parameter store

class ExpirableValue (value: ForwardRef('str | bytes | dict[str, Any]'), ttl: ForwardRef('datetime'))

ExpirableValue(value, ttl)

Expand source code
class ExpirableValue(NamedTuple):
    value: str | bytes | dict[str, Any]
    ttl: datetime

Ancestors

  • builtins.tuple

Instance variables

var ttl : datetime.datetime

Alias for field number 1

var value : str | bytes | dict[str, typing.Any]

Alias for field number 0