Module aws_lambda_powertools.utilities.parameters

Parameter retrieval and caching utility

# -*- coding: utf-8 -*-

"""
Parameter retrieval and caching utility
"""

from .appconfig import AppConfigProvider, get_app_config
from .base import BaseProvider
from .dynamodb import DynamoDBProvider
from .exceptions import GetParameterError, TransformParameterError
from .secrets import SecretsProvider, get_secret
from .ssm import SSMProvider, get_parameter, get_parameters

__all__ = [
    "AppConfigProvider",
    "BaseProvider",
    "GetParameterError",
    "DynamoDBProvider",
    "SecretsProvider",
    "SSMProvider",
    "TransformParameterError",
    "get_app_config",
    "get_parameter",
    "get_parameters",
    "get_secret",
]

Sub-modules

aws_lambda_powertools.utilities.parameters.appconfig

AWS App Config configuration retrieval and caching utility

aws_lambda_powertools.utilities.parameters.base

Base for Parameter providers

aws_lambda_powertools.utilities.parameters.dynamodb

Amazon DynamoDB parameter retrieval and caching utility

aws_lambda_powertools.utilities.parameters.exceptions

Parameter retrieval exceptions

aws_lambda_powertools.utilities.parameters.secrets

AWS Secrets Manager parameter retrieval and caching utility

aws_lambda_powertools.utilities.parameters.ssm

AWS SSM Parameter retrieval and caching utility

Functions

def get_app_config(name: str, environment: str, application: Union[str, NoneType] = None, transform: Union[str, NoneType] = None, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]

Retrieve a configuration value from AWS App Config.

Parameters

name : str

Name of the configuration

environment : str

Environment of the configuration

application : str

Application of the configuration

transform : str, optional

Transforms the content from a JSON object ('json') or base64 binary string ('binary')

force_fetch : bool, optional

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

sdk_options : dict, optional

Dictionary of options that will be passed to the Parameter Store get_parameter 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.

Example

Retrieves the latest version of configuration value from App Config

>>> from aws_lambda_powertools.utilities.parameters import get_app_config
>>>
>>> value = get_app_config("my_config", environment="my_env", application="my_env")
>>>
>>> print(value)
My configuration value

Retrieves a confiugration value and decodes it using a JSON decoder

>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> value = get_app_config("my_config", environment="my_env", application="my_env", transform='json')
>>>
>>> print(value)
My configuration's JSON value

Expand source code

def get_app_config(
    name: str,
    environment: str,
    application: Optional[str] = None,
    transform: Optional[str] = None,
    force_fetch: bool = False,
    **sdk_options
) -> Union[str, list, dict, bytes]:
    """
    Retrieve a configuration value from AWS App Config.

    Parameters
    ----------
    name: str
        Name of the configuration
    environment: str
        Environment of the configuration
    application: str
        Application of the configuration
    transform: str, optional
        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Dictionary of options that will be passed to the Parameter Store get_parameter 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.

    Example
    -------
    **Retrieves the latest version of configuration value from App Config**

        >>> from aws_lambda_powertools.utilities.parameters import get_app_config
        >>>
        >>> value = get_app_config("my_config", environment="my_env", application="my_env")
        >>>
        >>> print(value)
        My configuration value

    **Retrieves a confiugration value and decodes it using a JSON decoder**

        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
        >>>
        >>> value = get_app_config("my_config", environment="my_env", application="my_env", transform='json')
        >>>
        >>> print(value)
        My configuration's JSON value
    """

    # Only create the provider if this function is called at least once
    if "appconfig" not in DEFAULT_PROVIDERS:
        DEFAULT_PROVIDERS["appconfig"] = AppConfigProvider(environment=environment, application=application)

    sdk_options["ClientId"] = CLIENT_ID

    return DEFAULT_PROVIDERS["appconfig"].get(name, transform=transform, force_fetch=force_fetch, **sdk_options)

def get_parameter(name: str, transform: Union[str, NoneType] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]

Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store

Parameters

name : str

Name of the parameter

transform : str, optional

Transforms the content from a JSON object ('json') or base64 binary string ('binary')

decrypt : bool, optional

If the parameter values should be decrypted

force_fetch : bool, optional

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

sdk_options : dict, optional

Dictionary of options that will be passed to the Parameter Store get_parameter 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.

Example

Retrieves a parameter value from Systems Manager Parameter Store

>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> value = get_parameter("/my/parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value and decodes it using a Base64 decoder

>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> value = get_parameter("/my/parameter", transform='binary')
>>>
>>> print(value)
My parameter value

Expand source code

def get_parameter(
    name: str, transform: Optional[str] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options
) -> Union[str, list, dict, bytes]:
    """
    Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store

    Parameters
    ----------
    name: str
        Name of the parameter
    transform: str, optional
        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
    decrypt: bool, optional
        If the parameter values should be decrypted
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Dictionary of options that will be passed to the Parameter Store get_parameter 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.

    Example
    -------
    **Retrieves a parameter value from Systems Manager Parameter Store**

        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
        >>>
        >>> value = get_parameter("/my/parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value and decodes it using a Base64 decoder**

        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
        >>>
        >>> value = get_parameter("/my/parameter", transform='binary')
        >>>
        >>> print(value)
        My parameter value
    """

    # Only create the provider if this function is called at least once
    if "ssm" not in DEFAULT_PROVIDERS:
        DEFAULT_PROVIDERS["ssm"] = SSMProvider()

    # Add to `decrypt` sdk_options to we can have an explicit option for this
    sdk_options["decrypt"] = decrypt

    return DEFAULT_PROVIDERS["ssm"].get(name, transform=transform, force_fetch=force_fetch, **sdk_options)

def get_parameters(path: str, transform: Union[str, NoneType] = None, recursive: bool = True, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]

Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store

Parameters

path : str

Path to retrieve the parameters

transform : str, optional

Transforms the content from a JSON object ('json') or base64 binary string ('binary')

recursive : bool, optional

If this should retrieve the parameter values recursively or not, defaults to True

decrypt : bool, optional

If the parameter values should be decrypted

force_fetch : bool, optional

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

sdk_options : dict, optional

Dictionary of options that will be passed to the Parameter Store get_parameters_by_path 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.

Example

Retrieves parameter values from Systems Manager Parameter Store

>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> values = get_parameters("/my/path/prefix")
>>>
>>> for key, value in values.items():
...     print(key, value)
/my/path/prefix/a   Parameter value a
/my/path/prefix/b   Parameter value b
/my/path/prefix/c   Parameter value c

Retrieves parameter values and decodes them using a Base64 decoder

>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> values = get_parameters("/my/path/prefix", transform='binary')

Expand source code

def get_parameters(
    path: str,
    transform: Optional[str] = None,
    recursive: bool = True,
    decrypt: bool = False,
    force_fetch: bool = False,
    **sdk_options
) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
    """
    Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store

    Parameters
    ----------
    path: str
        Path to retrieve the parameters
    transform: str, optional
        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
    recursive: bool, optional
        If this should retrieve the parameter values recursively or not, defaults to True
    decrypt: bool, optional
        If the parameter values should be decrypted
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Dictionary of options that will be passed to the Parameter Store get_parameters_by_path 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.

    Example
    -------
    **Retrieves parameter values from Systems Manager Parameter Store**

        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
        >>>
        >>> values = get_parameters("/my/path/prefix")
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        /my/path/prefix/a   Parameter value a
        /my/path/prefix/b   Parameter value b
        /my/path/prefix/c   Parameter value c

    **Retrieves parameter values and decodes them using a Base64 decoder**

        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
        >>>
        >>> values = get_parameters("/my/path/prefix", transform='binary')
    """

    # Only create the provider if this function is called at least once
    if "ssm" not in DEFAULT_PROVIDERS:
        DEFAULT_PROVIDERS["ssm"] = SSMProvider()

    sdk_options["recursive"] = recursive
    sdk_options["decrypt"] = decrypt

    return DEFAULT_PROVIDERS["ssm"].get_multiple(path, transform=transform, force_fetch=force_fetch, **sdk_options)

def get_secret(name: str, transform: Union[str, NoneType] = None, force_fetch: bool = False, **sdk_options) ‑> Union[str, dict, bytes]

Retrieve a parameter value from AWS Secrets Manager

Parameters

name : str

Name of the parameter

transform : str, optional

Transforms the content from a JSON object ('json') or base64 binary string ('binary')

force_fetch : bool, optional

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

sdk_options : dict, optional

Dictionary of options that will be passed to the get_secret_value 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.

Example

Retrieves a secret*

>>> from aws_lambda_powertools.utilities.parameters import get_secret
>>>
>>> get_secret("my-secret")

Retrieves a secret and transforms using a JSON deserializer*

>>> from aws_lambda_powertools.utilities.parameters import get_secret
>>>
>>> get_secret("my-secret", transform="json")

Retrieves a secret and passes custom arguments to the SDK

>>> from aws_lambda_powertools.utilities.parameters import get_secret
>>>
>>> get_secret("my-secret", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")

Expand source code

def get_secret(
    name: str, transform: Optional[str] = None, force_fetch: bool = False, **sdk_options
) -> Union[str, dict, bytes]:
    """
    Retrieve a parameter value from AWS Secrets Manager

    Parameters
    ----------
    name: str
        Name of the parameter
    transform: str, optional
        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
    force_fetch: bool, optional
        Force update even before a cached item has expired, defaults to False
    sdk_options: dict, optional
        Dictionary of options that will be passed to the get_secret_value 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.

    Example
    -------
    **Retrieves a secret***

        >>> from aws_lambda_powertools.utilities.parameters import get_secret
        >>>
        >>> get_secret("my-secret")

    **Retrieves a secret and transforms using a JSON deserializer***

        >>> from aws_lambda_powertools.utilities.parameters import get_secret
        >>>
        >>> get_secret("my-secret", transform="json")

    **Retrieves a secret and passes custom arguments to the SDK**

        >>> from aws_lambda_powertools.utilities.parameters import get_secret
        >>>
        >>> get_secret("my-secret", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
    """

    # Only create the provider if this function is called at least once
    if "secrets" not in DEFAULT_PROVIDERS:
        DEFAULT_PROVIDERS["secrets"] = SecretsProvider()

    return DEFAULT_PROVIDERS["secrets"].get(name, transform=transform, force_fetch=force_fetch, **sdk_options)

Classes

class AppConfigProvider (environment: str, application: Union[str, NoneType] = None, config: Union[botocore.config.Config, NoneType] = None)

AWS App Config Provider

Parameters

environment : str

Environment of the configuration to pass during client initialization

application : str, optional

Application of the configuration to pass during client initialization

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

Example

Retrieves the latest configuration value from App Config

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app")
>>>
>>> value : bytes = appconf_provider.get("my_conf")
>>>
>>> print(value)
My configuration value

Retrieves a configuration value from App Config in another AWS region

>>> from botocore.config import Config
>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> config = Config(region_name="us-west-1")
>>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app", config=config)
>>>
>>> value : bytes = appconf_provider.get("my_conf")
>>>
>>> print(value)
My configuration value

Initialize the App Config client

Expand source code

class AppConfigProvider(BaseProvider):
    """
    AWS App Config Provider

    Parameters
    ----------
    environment: str
        Environment of the configuration to pass during client initialization
    application: str, optional
        Application of the configuration to pass during client initialization
    config: botocore.config.Config, optional
        Botocore configuration to pass during client initialization

    Example
    -------
    **Retrieves the latest configuration value from App Config**

        >>> from aws_lambda_powertools.utilities import parameters
        >>>
        >>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app")
        >>>
        >>> value : bytes = appconf_provider.get("my_conf")
        >>>
        >>> print(value)
        My configuration value

    **Retrieves a configuration value from App Config in another AWS region**

        >>> from botocore.config import Config
        >>> from aws_lambda_powertools.utilities import parameters
        >>>
        >>> config = Config(region_name="us-west-1")
        >>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app", config=config)
        >>>
        >>> value : bytes = appconf_provider.get("my_conf")
        >>>
        >>> print(value)
        My configuration value

    """

    client: Any = None

    def __init__(self, environment: str, application: Optional[str] = None, config: Optional[Config] = None):
        """
        Initialize the App Config client
        """

        config = config or Config()
        self.client = boto3.client("appconfig", config=config)
        self.application = resolve_env_var_choice(
            choice=application, env=os.getenv(constants.SERVICE_NAME_ENV, "service_undefined")
        )
        self.environment = environment
        self.current_version = ""

        super().__init__()

    def _get(self, name: str, **sdk_options) -> str:
        """
        Retrieve a parameter value from AWS App config.

        Parameters
        ----------
        name: str
            Name of the configuration
        environment: str
            Environment of the configuration
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Parameter Store get_parameter API call
        """

        sdk_options["Configuration"] = name
        sdk_options["Application"] = self.application
        sdk_options["Environment"] = self.environment
        sdk_options["ClientId"] = CLIENT_ID

        response = self.client.get_configuration(**sdk_options)
        return response["Content"].read()  # read() of botocore.response.StreamingBody

    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
        """
        Retrieving multiple parameter values is not supported with AWS App Config Provider
        """
        raise NotImplementedError()

Ancestors

• BaseProvider
• abc.ABC

Class variables

var client : Any

Inherited members

• BaseProvider:
  • get
  • get_multiple

class BaseProvider

Abstract Base Class for Parameter providers

Initialize the base provider

Expand source code

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

    store: Any = None

    def __init__(self):
        """
        Initialize the base provider
        """

        self.store = {}

    def _has_not_expired(self, key: Tuple[str, Optional[str]]) -> bool:
        return key in self.store and self.store[key].ttl >= datetime.now()

    def get(
        self,
        name: str,
        max_age: int = DEFAULT_MAX_AGE_SECS,
        transform: Optional[str] = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Union[str, list, dict, bytes]:
        """
        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.
        key = (name, transform)

        if not force_fetch and self._has_not_expired(key):
            return self.store[key].value

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

        if transform is not None:
            value = transform_value(value, transform)

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

        return value

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

    def get_multiple(
        self,
        path: str,
        max_age: int = DEFAULT_MAX_AGE_SECS,
        transform: Optional[str] = None,
        raise_on_transform_error: bool = False,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
        """
        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 = (path, transform)

        if not force_fetch and self._has_not_expired(key):
            return self.store[key].value

        try:
            values: Dict[str, Union[str, bytes, dict, None]] = self._get_multiple(path, **sdk_options)
        # Encapsulate all errors into a generic GetParameterError
        except Exception as exc:
            raise GetParameterError(str(exc))

        if transform is not None:
            for (key, value) in values.items():
                _transform = get_transform_method(key, transform)
                if _transform is None:
                    continue

                values[key] = transform_value(value, _transform, raise_on_transform_error)

        self.store[key] = ExpirableValue(values, datetime.now() + timedelta(seconds=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()

Ancestors

• abc.ABC

Subclasses

• AppConfigProvider
• DynamoDBProvider
• SecretsProvider
• SSMProvider

Class variables

var store : Any

Methods

def get(self, name: str, max_age: int = 5, transform: Union[str, NoneType] = None, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]

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.

Expand source code

def get(
    self,
    name: str,
    max_age: int = DEFAULT_MAX_AGE_SECS,
    transform: Optional[str] = None,
    force_fetch: bool = False,
    **sdk_options,
) -> Union[str, list, dict, bytes]:
    """
    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.
    key = (name, transform)

    if not force_fetch and self._has_not_expired(key):
        return self.store[key].value

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

    if transform is not None:
        value = transform_value(value, transform)

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

    return value

def get_multiple(self, path: str, max_age: int = 5, transform: Union[str, NoneType] = None, raise_on_transform_error: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]

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.

Expand source code

def get_multiple(
    self,
    path: str,
    max_age: int = DEFAULT_MAX_AGE_SECS,
    transform: Optional[str] = None,
    raise_on_transform_error: bool = False,
    force_fetch: bool = False,
    **sdk_options,
) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
    """
    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 = (path, transform)

    if not force_fetch and self._has_not_expired(key):
        return self.store[key].value

    try:
        values: Dict[str, Union[str, bytes, dict, None]] = self._get_multiple(path, **sdk_options)
    # Encapsulate all errors into a generic GetParameterError
    except Exception as exc:
        raise GetParameterError(str(exc))

    if transform is not None:
        for (key, value) in values.items():
            _transform = get_transform_method(key, transform)
            if _transform is None:
                continue

            values[key] = transform_value(value, _transform, raise_on_transform_error)

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

    return values

class DynamoDBProvider (table_name: str, key_attr: str = 'id', sort_attr: str = 'sk', value_attr: str = 'value', endpoint_url: Union[str, NoneType] = None, config: Union[botocore.config.Config, NoneType] = None)

Amazon DynamoDB Parameter Provider

Parameters

table_name : str

Name of the DynamoDB table that stores parameters

key_attr : str, optional

Hash key for the DynamoDB table (default to 'id')

sort_attr : str, optional

Name of the DynamoDB table sort key (defaults to 'sk'), used only for get_multiple

value_attr : str, optional

Attribute that contains the values in the DynamoDB table (defaults to 'value')

endpoint_url : str, optional

Complete url to reference local DynamoDB instance, e.g. http://localhost:8080

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

Example

Retrieves a parameter value from a DynamoDB table

In this example, the DynamoDB table uses id as hash key and stores the value in the value attribute. The parameter item looks like this:

{ "id": "my-parameters", "value": "Parameter value a" }

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider("ParametersTable")
>>>
>>> value = ddb_provider.get("my-parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from a DynamoDB table that has custom attribute names

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider(
...     "ParametersTable",
...     key_attr="my-id",
...     value_attr="my-value"
... )
>>>
>>> value = ddb_provider.get("my-parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from a DynamoDB table in another AWS region

>>> from botocore.config import Config
>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>>
>>> config = Config(region_name="us-west-1")
>>> ddb_provider = DynamoDBProvider("ParametersTable", config=config)
>>>
>>> value = ddb_provider.get("my-parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from a DynamoDB table passing options to the SDK call

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider("ParametersTable")
>>>
>>> value = ddb_provider.get("my-parameter", ConsistentRead=True)
>>>
>>> print(value)
My parameter value

Retrieves multiple values from a DynamoDB table

In this case, the provider will use a sort key to retrieve multiple values using a query under the hood. This expects that the sort key is named sk. The DynamoDB table contains three items looking like this:

{ "id": "my-parameters", "sk": "a", "value": "Parameter value a" }
{ "id": "my-parameters", "sk": "b", "value": "Parameter value b" }
{ "id": "my-parameters", "sk": "c", "value": "Parameter value c" }

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider("ParametersTable")
>>>
>>> values = ddb_provider.get_multiple("my-parameters")
>>>
>>> for key, value in values.items():
...     print(key, value)
a   Parameter value a
b   Parameter value b
c   Parameter value c

Retrieves multiple values from a DynamoDB table that has custom attribute names

In this case, the provider will use a sort key to retrieve multiple values using a query under the hood.

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider(
...     "ParametersTable",
...     key_attr="my-id",
...     sort_attr="my-sort-key",
...     value_attr="my-value"
... )
>>>
>>> values = ddb_provider.get_multiple("my-parameters")
>>>
>>> for key, value in values.items():
...     print(key, value)
a   Parameter value a
b   Parameter value b
c   Parameter value c

Retrieves multiple values from a DynamoDB table passing options to the SDK calls

>>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
>>> ddb_provider = DynamoDBProvider("ParametersTable")
>>>
>>> values = ddb_provider.get_multiple("my-parameters", ConsistentRead=True)
>>>
>>> for key, value in values.items():
...     print(key, value)
a   Parameter value a
b   Parameter value b
c   Parameter value c

Initialize the DynamoDB client

Expand source code

class DynamoDBProvider(BaseProvider):
    """
    Amazon DynamoDB Parameter Provider

    Parameters
    ----------
    table_name: str
        Name of the DynamoDB table that stores parameters
    key_attr: str, optional
        Hash key for the DynamoDB table (default to 'id')
    sort_attr: str, optional
        Name of the DynamoDB table sort key (defaults to 'sk'), used only for get_multiple
    value_attr: str, optional
        Attribute that contains the values in the DynamoDB table (defaults to 'value')
    endpoint_url: str, optional
        Complete url to reference local DynamoDB instance, e.g. http://localhost:8080
    config: botocore.config.Config, optional
        Botocore configuration to pass during client initialization

    Example
    -------
    **Retrieves a parameter value from a DynamoDB table**

    In this example, the DynamoDB table uses `id` as hash key and stores the value in the `value`
    attribute. The parameter item looks like this:

        { "id": "my-parameters", "value": "Parameter value a" }

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider("ParametersTable")
        >>>
        >>> value = ddb_provider.get("my-parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from a DynamoDB table that has custom attribute names**

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider(
        ...     "ParametersTable",
        ...     key_attr="my-id",
        ...     value_attr="my-value"
        ... )
        >>>
        >>> value = ddb_provider.get("my-parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from a DynamoDB table in another AWS region**

        >>> from botocore.config import Config
        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>>
        >>> config = Config(region_name="us-west-1")
        >>> ddb_provider = DynamoDBProvider("ParametersTable", config=config)
        >>>
        >>> value = ddb_provider.get("my-parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from a DynamoDB table passing options to the SDK call**

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider("ParametersTable")
        >>>
        >>> value = ddb_provider.get("my-parameter", ConsistentRead=True)
        >>>
        >>> print(value)
        My parameter value

    **Retrieves multiple values from a DynamoDB table**

    In this case, the provider will use a sort key to retrieve multiple values using a query under
    the hood. This expects that the sort key is named `sk`. The DynamoDB table contains three items
    looking like this:

        { "id": "my-parameters", "sk": "a", "value": "Parameter value a" }
        { "id": "my-parameters", "sk": "b", "value": "Parameter value b" }
        { "id": "my-parameters", "sk": "c", "value": "Parameter value c" }

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider("ParametersTable")
        >>>
        >>> values = ddb_provider.get_multiple("my-parameters")
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        a   Parameter value a
        b   Parameter value b
        c   Parameter value c

    **Retrieves multiple values from a DynamoDB table that has custom attribute names**

    In this case, the provider will use a sort key to retrieve multiple values using a query under
    the hood.

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider(
        ...     "ParametersTable",
        ...     key_attr="my-id",
        ...     sort_attr="my-sort-key",
        ...     value_attr="my-value"
        ... )
        >>>
        >>> values = ddb_provider.get_multiple("my-parameters")
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        a   Parameter value a
        b   Parameter value b
        c   Parameter value c

    **Retrieves multiple values from a DynamoDB table passing options to the SDK calls**

        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
        >>> ddb_provider = DynamoDBProvider("ParametersTable")
        >>>
        >>> values = ddb_provider.get_multiple("my-parameters", ConsistentRead=True)
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        a   Parameter value a
        b   Parameter value b
        c   Parameter value c
    """

    table: Any = None
    key_attr = None
    sort_attr = None
    value_attr = None

    def __init__(
        self,
        table_name: str,
        key_attr: str = "id",
        sort_attr: str = "sk",
        value_attr: str = "value",
        endpoint_url: Optional[str] = None,
        config: Optional[Config] = None,
    ):
        """
        Initialize the DynamoDB client
        """

        config = config or Config()
        self.table = boto3.resource("dynamodb", endpoint_url=endpoint_url, config=config).Table(table_name)

        self.key_attr = key_attr
        self.sort_attr = sort_attr
        self.value_attr = value_attr

        super().__init__()

    def _get(self, name: str, **sdk_options) -> str:
        """
        Retrieve a parameter value from Amazon DynamoDB

        Parameters
        ----------
        name: str
            Name of the parameter
        sdk_options: dict, optional
            Dictionary of options that will be passed to the DynamoDB get_item API call
        """

        # Explicit arguments will take precedence over keyword arguments
        sdk_options["Key"] = {self.key_attr: name}

        return self.table.get_item(**sdk_options)["Item"][self.value_attr]

    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
        """
        Retrieve multiple parameter values from Amazon DynamoDB

        Parameters
        ----------
        path: str
            Path to retrieve the parameters
        sdk_options: dict, optional
            Dictionary of options that will be passed to the DynamoDB query API call
        """

        # Explicit arguments will take precedence over keyword arguments
        sdk_options["KeyConditionExpression"] = Key(self.key_attr).eq(path)

        response = self.table.query(**sdk_options)
        items = response.get("Items", [])

        # Keep querying while there are more items matching the partition key
        while "LastEvaluatedKey" in response:
            sdk_options["ExclusiveStartKey"] = response["LastEvaluatedKey"]
            response = self.table.query(**sdk_options)
            items.extend(response.get("Items", []))

        return {item[self.sort_attr]: item[self.value_attr] for item in items}

Ancestors

• BaseProvider
• abc.ABC

Class variables

var key_attr

var sort_attr

var table : Any

var value_attr

Inherited members

• BaseProvider:
  • get
  • get_multiple

class GetParameterError (*args, **kwargs)

When a provider raises an exception on parameter retrieval

Expand source code

class GetParameterError(Exception):
    """When a provider raises an exception on parameter retrieval"""

Ancestors

• builtins.Exception
• builtins.BaseException

class SSMProvider (config: Union[botocore.config.Config, NoneType] = None)

AWS Systems Manager Parameter Store Provider

Parameters

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

Example

Retrieves a parameter value from Systems Manager Parameter Store

>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> value = ssm_provider.get("/my/parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from Systems Manager Parameter Store in another AWS region

>>> from botocore.config import Config
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>>
>>> config = Config(region_name="us-west-1")
>>> ssm_provider = SSMProvider(config=config)
>>>
>>> value = ssm_provider.get("/my/parameter")
>>>
>>> print(value)
My parameter value

Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix

>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> values = ssm_provider.get_multiple("/my/path/prefix")
>>>
>>> for key, value in values.items():
...     print(key, value)
/my/path/prefix/a   Parameter value a
/my/path/prefix/b   Parameter value b
/my/path/prefix/c   Parameter value c

Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call

>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10)
>>>
>>> for key, value in values.items():
...     print(key, value)
/my/path/prefix/a   Parameter value a
/my/path/prefix/b   Parameter value b
/my/path/prefix/c   Parameter value c

Initialize the SSM Parameter Store client

Expand source code

class SSMProvider(BaseProvider):
    """
    AWS Systems Manager Parameter Store Provider

    Parameters
    ----------
    config: botocore.config.Config, optional
        Botocore configuration to pass during client initialization

    Example
    -------
    **Retrieves a parameter value from Systems Manager Parameter Store**

        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
        >>> ssm_provider = SSMProvider()
        >>>
        >>> value = ssm_provider.get("/my/parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from Systems Manager Parameter Store in another AWS region**

        >>> from botocore.config import Config
        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
        >>>
        >>> config = Config(region_name="us-west-1")
        >>> ssm_provider = SSMProvider(config=config)
        >>>
        >>> value = ssm_provider.get("/my/parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix**

        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
        >>> ssm_provider = SSMProvider()
        >>>
        >>> values = ssm_provider.get_multiple("/my/path/prefix")
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        /my/path/prefix/a   Parameter value a
        /my/path/prefix/b   Parameter value b
        /my/path/prefix/c   Parameter value c

    **Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call**

        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
        >>> ssm_provider = SSMProvider()
        >>>
        >>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10)
        >>>
        >>> for key, value in values.items():
        ...     print(key, value)
        /my/path/prefix/a   Parameter value a
        /my/path/prefix/b   Parameter value b
        /my/path/prefix/c   Parameter value c
    """

    client: Any = None

    def __init__(self, config: Optional[Config] = None):
        """
        Initialize the SSM Parameter Store client
        """

        config = config or Config()
        self.client = boto3.client("ssm", config=config)

        super().__init__()

    def get(
        self,
        name: str,
        max_age: int = DEFAULT_MAX_AGE_SECS,
        transform: Optional[str] = None,
        decrypt: bool = False,
        force_fetch: bool = False,
        **sdk_options
    ) -> Union[str, list, dict, bytes]:
        """
        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.
        decrypt: bool, optional
            If the parameter value should be decrypted
        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.
        """

        # Add to `decrypt` sdk_options to we can have an explicit option for this
        sdk_options["decrypt"] = decrypt

        return super().get(name, max_age, transform, force_fetch, **sdk_options)

    def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str:
        """
        Retrieve a parameter value from AWS Systems Manager Parameter Store

        Parameters
        ----------
        name: str
            Parameter name
        decrypt: bool, optional
            If the parameter value should be decrypted
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Parameter Store get_parameter API call
        """

        # Explicit arguments will take precedence over keyword arguments
        sdk_options["Name"] = name
        sdk_options["WithDecryption"] = decrypt

        return self.client.get_parameter(**sdk_options)["Parameter"]["Value"]

    def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = False, **sdk_options) -> Dict[str, str]:
        """
        Retrieve multiple parameter values from AWS Systems Manager Parameter Store

        Parameters
        ----------
        path: str
            Path to retrieve the parameters
        decrypt: bool, optional
            If the parameter values should be decrypted
        recursive: bool, optional
            If this should retrieve the parameter values recursively or not
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call
        """

        # Explicit arguments will take precedence over keyword arguments
        sdk_options["Path"] = path
        sdk_options["WithDecryption"] = decrypt
        sdk_options["Recursive"] = recursive

        parameters = {}
        for page in self.client.get_paginator("get_parameters_by_path").paginate(**sdk_options):
            for parameter in page.get("Parameters", []):
                # Standardize the parameter name
                # The parameter name returned by SSM will contained the full path.
                # However, for readability, we should return only the part after
                # the path.
                name = parameter["Name"]
                if name.startswith(path):
                    name = name[len(path) :]
                name = name.lstrip("/")

                parameters[name] = parameter["Value"]

        return parameters

Ancestors

• BaseProvider
• abc.ABC

Class variables

var client : Any

Methods

def get(self, name: str, max_age: int = 5, transform: Union[str, NoneType] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]

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.

decrypt : bool, optional

If the parameter value should be decrypted

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.

Expand source code

def get(
    self,
    name: str,
    max_age: int = DEFAULT_MAX_AGE_SECS,
    transform: Optional[str] = None,
    decrypt: bool = False,
    force_fetch: bool = False,
    **sdk_options
) -> Union[str, list, dict, bytes]:
    """
    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.
    decrypt: bool, optional
        If the parameter value should be decrypted
    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.
    """

    # Add to `decrypt` sdk_options to we can have an explicit option for this
    sdk_options["decrypt"] = decrypt

    return super().get(name, max_age, transform, force_fetch, **sdk_options)

Inherited members

• BaseProvider:
  • get_multiple

class SecretsProvider (config: Union[botocore.config.Config, NoneType] = None)

AWS Secrets Manager Parameter Provider

Parameters

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

Example

Retrieves a parameter value from Secrets Manager

>>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
>>> secrets_provider = SecretsProvider()
>>>
>>> value = secrets_provider.get("my-parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from Secrets Manager in another AWS region

>>> from botocore.config import Config
>>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
>>>
>>> config = Config(region_name="us-west-1")
>>> secrets_provider = SecretsProvider(config=config)
>>>
>>> value = secrets_provider.get("my-parameter")
>>>
>>> print(value)
My parameter value

Retrieves a parameter value from Secrets Manager passing options to the SDK call

>>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
>>> secrets_provider = SecretsProvider()
>>>
>>> value = secrets_provider.get("my-parameter", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
>>>
>>> print(value)
My parameter value

Initialize the Secrets Manager client

Expand source code

class SecretsProvider(BaseProvider):
    """
    AWS Secrets Manager Parameter Provider

    Parameters
    ----------
    config: botocore.config.Config, optional
        Botocore configuration to pass during client initialization

    Example
    -------
    **Retrieves a parameter value from Secrets Manager**

        >>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
        >>> secrets_provider = SecretsProvider()
        >>>
        >>> value = secrets_provider.get("my-parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from Secrets Manager in another AWS region**

        >>> from botocore.config import Config
        >>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
        >>>
        >>> config = Config(region_name="us-west-1")
        >>> secrets_provider = SecretsProvider(config=config)
        >>>
        >>> value = secrets_provider.get("my-parameter")
        >>>
        >>> print(value)
        My parameter value

    **Retrieves a parameter value from Secrets Manager passing options to the SDK call**

        >>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
        >>> secrets_provider = SecretsProvider()
        >>>
        >>> value = secrets_provider.get("my-parameter", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
        >>>
        >>> print(value)
        My parameter value
    """

    client: Any = None

    def __init__(self, config: Optional[Config] = None):
        """
        Initialize the Secrets Manager client
        """

        config = config or Config()

        self.client = boto3.client("secretsmanager", config=config)

        super().__init__()

    def _get(self, name: str, **sdk_options) -> str:
        """
        Retrieve a parameter value from AWS Systems Manager Parameter Store

        Parameters
        ----------
        name: str
            Name of the parameter
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Secrets Manager get_secret_value API call
        """

        # Explicit arguments will take precedence over keyword arguments
        sdk_options["SecretId"] = name

        return self.client.get_secret_value(**sdk_options)["SecretString"]

    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
        """
        Retrieving multiple parameter values is not supported with AWS Secrets Manager
        """
        raise NotImplementedError()

Ancestors

• BaseProvider
• abc.ABC

Class variables

var client : Any

Inherited members

• BaseProvider:
  • get
  • get_multiple

class TransformParameterError (*args, **kwargs)

When a provider fails to transform a parameter value

Expand source code

class TransformParameterError(Exception):
    """When a provider fails to transform a parameter value"""

Ancestors

• builtins.Exception
• builtins.BaseException