Module aws_lambda_powertools.utilities.parameters

Parameter retrieval and caching utility

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

aws_lambda_powertools.utilities.parameters.types

Functions

def clear_caches()

Clear cached parameter values from all providers

def get_app_config(name: str, environment: str, application: Optional[str] = None, transform: Literal['json', 'binary', 'auto', None] = None, force_fetch: bool = False, max_age: Optional[int] = None, **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

max_age : int, optional

Maximum age of the cached value

sdk_options : dict, optional

SDK options to propagate to start_configuration_session 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 configuration value and decodes it using a JSON decoder

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

def get_parameter(name: str, transform: TransformOptions = None, decrypt: Optional[bool] = None, force_fetch: bool = False, max_age: Optional[int] = None, **sdk_options) ‑> Union[str, 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

max_age : int, optional

Maximum age of the cached value

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

def get_parameters(path: str, transform: TransformOptions = None, recursive: bool = True, decrypt: Optional[bool] = None, force_fetch: bool = False, max_age: Optional[int] = None, raise_on_transform_error: 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

For readability, we strip the path prefix name in the response.

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

max_age : int, optional

Maximum age of the cached value

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

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)
config              Parameter value (/my/path/prefix/config)
webhook/config      Parameter value (/my/path/prefix/webhook/config)

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

def get_parameters_by_name(parameters: Dict[str, Any], transform: TransformOptions = None, decrypt: Optional[bool] = None, max_age: Optional[int] = None, raise_on_error: bool = True) ‑> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]

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

Parameters

parameters : List[Dict[str, Dict]]

List of parameter names, and any optional overrides

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

max_age : int, optional

Maximum age of the cached value

raise_on_error : bool, optional

Whether to fail-fast or fail gracefully by including "_errors" key in the response, by default True

Example

Retrieves multiple parameters from distinct paths from Systems Manager Parameter Store

from aws_lambda_powertools.utilities.parameters import get_parameters_by_name

params = {
    "/param": {},
    "/json": {"transform": "json"},
    "/binary": {"transform": "binary"},
    "/no_cache": {"max_age": 0},
    "/api_key": {"decrypt": True},
}

values = get_parameters_by_name(parameters=params)
for param_name, value in values.items():
    print(f"{param_name}: {value}")

# "/param": value
# "/json": value
# "/binary": value
# "/no_cache": value
# "/api_key": value

Raises

GetParameterError

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

def get_secret(name: str, transform: TransformOptions = None, force_fetch: bool = False, max_age: Optional[int] = None, **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

max_age : int, optional

Maximum age of the cached value

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

def set_parameter(name: str, value: str, *, overwrite: bool = False, description: str = '', parameter_type: SSM_PARAMETER_TYPES = 'String', tier: SSM_PARAMETER_TIER = 'Standard', kms_key_id: str | None = None, **sdk_options) ‑> PutParameterResponse

Sets a parameter in AWS Systems Manager Parameter Store.

Parameters

name : str

The fully qualified name includes the complete hierarchy of the parameter name and name.

value : str

The parameter value

overwrite : bool, optional

If the parameter value should be overwritten, False by default

description : str, optional

The description of the parameter

parameter_type : str, optional

Type of the parameter. Allowed values are String, StringList, and SecureString

tier : str, optional

The parameter tier to use. Allowed values are Standard, Advanced, and Intelligent-Tiering

kms_key_id : str, optional

The KMS key id to use to encrypt the parameter

sdk_options : dict, optional

Dictionary of options that will be passed to the Parameter Store get_parameter API call

Raises

SetParameterError

When attempting to set a parameter fails.

URLs:

<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html>

Example

Sets a parameter value from Systems Manager Parameter Store

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> response = parameters.set_parameter(name="/my/example/parameter", value="More Powertools")
>>>
>>> print(response)
123

Returns

PutParameterResponse

The dict returned by boto3.

def set_secret(name: str, value: Union[str, bytes], *, client_request_token: Optional[str] = None, **sdk_options) ‑> SetSecretResponse

Modify the details of a secret or create a new secret if it doesn't already exist.

We aim to minimize API calls by assuming that the secret already exists and needs updating. If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding:

              ┌────────────────────────┐      ┌─────────────────┐
    ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐
    │         └────────────────────────┘      └─────────────────┘     │
    │                                                                 │
    │                                                                 │
    │                                                                 ▼

┌─────────────────┐ ┌─────────────────────┐ │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │ └─────────────────┘ └─────────────────────┘

Parameters

name : str

The ARN or name of the secret to add a new version to or create a new one.

value : str, dict or bytes

Specifies text data that you want to encrypt and store in this new version of the secret.

client_request_token : str, optional

This value helps ensure idempotency. It's recommended that you generate a UUID-type value to ensure uniqueness within the specified secret. This value becomes the VersionId of the new version. This field is auto-populated if not provided, but no idempotency will be enforced this way.

sdk_options : dict, optional

Dictionary of options that will be passed to the Secrets Manager update_secret API call

Raises

SetSecretError

When attempting to update or create a secret fails.

Returns:

Setsecretresponse

The dict returned by boto3.

Example

Sets a secret*

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd")

Sets a secret and includes an client_request_token

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> parameters.set_secret(
        name="my-secret",
        value='{"password": "supers3cr3tllam@passw0rd"}',
        client_request_token="YOUR_TOKEN_HERE"
    )

URLs:

<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html>
<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html>

Classes

class AppConfigProvider (environment: str, application: Optional[str] = None, config: Optional[botocore.config.Config] = None, boto3_session: Optional[boto3.session.Session] = None, boto3_client: Optional[ForwardRef('AppConfigDataClient')] = 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

boto3_session : boto3.session.Session, optional

Boto3 session to create a boto3_client from

boto3_client : AppConfigDataClient, optional

Boto3 AppConfigData Client to use, boto3_session will be ignored if both are provided

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
    boto3_session : boto3.session.Session, optional
            Boto3 session to create a boto3_client from
    boto3_client: AppConfigDataClient, optional
            Boto3 AppConfigData Client to use, boto3_session will be ignored if both are provided

    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,
        boto3_session: Optional[boto3.session.Session] = None,
        boto3_client: Optional["AppConfigDataClient"] = None,
    ):
        """
        Initialize the App Config client
        """

        super().__init__()

        self.client: "AppConfigDataClient" = self._build_boto3_client(
            service_name="appconfigdata",
            client=boto3_client,
            session=boto3_session,
            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 = ""

        self._next_token: Dict[str, str] = {}  # nosec - token for get_latest_configuration executions
        # Dict to store the recently retrieved value for a specific configuration.
        self.last_returned_value: Dict[str, str] = {}

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

        Parameters
        ----------
        name: str
            Name of the configuration
        sdk_options: dict, optional
            SDK options to propagate to `start_configuration_session` API call
        """
        if name not in self._next_token:
            sdk_options["ConfigurationProfileIdentifier"] = name
            sdk_options["ApplicationIdentifier"] = self.application
            sdk_options["EnvironmentIdentifier"] = self.environment
            response_configuration = self.client.start_configuration_session(**sdk_options)
            self._next_token[name] = response_configuration["InitialConfigurationToken"]

        # The new AppConfig APIs require two API calls to return the configuration
        # First we start the session and after that we retrieve the configuration
        # We need to store the token to use in the next execution
        response = self.client.get_latest_configuration(ConfigurationToken=self._next_token[name])
        return_value = response["Configuration"].read()
        self._next_token[name] = response["NextPollConfigurationToken"]

        # The return of get_latest_configuration can be null because this value is supposed to be cached
        # on the customer side.
        # We created a dictionary that stores the most recently retrieved value for a specific configuration.
        # See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata/client/get_latest_configuration.html
        if return_value:
            self.last_returned_value[name] = return_value

        return self.last_returned_value[name]

    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
  • set

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: Dict[Tuple, ExpirableValue]

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

        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: Optional[int] = None,
        transform: TransformOptions = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Optional[Union[str, 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.
        value: Optional[Union[str, bytes, dict]] = 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) -> Union[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: Optional[int] = None,
        transform: TransformOptions = 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 = 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)

    @staticmethod
    def _build_boto3_client(
        service_name: str,
        client: Optional[ParameterClients] = None,
        session: Optional[Type[boto3.Session]] = None,
        config: Optional[Type[Config]] = None,
    ) -> Type[ParameterClients]:
        """Builds a low level boto3 client with session and config provided

        Parameters
        ----------
        service_name : str
            AWS service name to instantiate a boto3 client, e.g. ssm
        client : Optional[ParameterClients], optional
            boto3 client instance, by default None
        session : Optional[Type[boto3.Session]], optional
            boto3 session instance, by default None
        config : Optional[Type[Config]], optional
            botocore config instance to configure client with, by default None

        Returns
        -------
        Type[ParameterClients]
            Instance of a boto3 client for Parameters feature (e.g., ssm, appconfig, secretsmanager, etc.)
        """
        if client is not None:
            user_agent.register_feature_to_client(client=client, feature="parameters")
            return client

        session = session or boto3.Session()
        config = config or Config()
        client = session.client(service_name=service_name, config=config)
        user_agent.register_feature_to_client(client=client, feature="parameters")
        return client

    # maintenance: change DynamoDBServiceResource type to ParameterResourceClients when we expand
    @staticmethod
    def _build_boto3_resource_client(
        service_name: str,
        client: Optional["DynamoDBServiceResource"] = None,
        session: Optional[Type[boto3.Session]] = None,
        config: Optional[Type[Config]] = None,
        endpoint_url: Optional[str] = None,
    ) -> "DynamoDBServiceResource":
        """Builds a high level boto3 resource client with session, config and endpoint_url provided

        Parameters
        ----------
        service_name : str
            AWS service name to instantiate a boto3 client, e.g. ssm
        client : Optional[DynamoDBServiceResource], optional
            boto3 client instance, by default None
        session : Optional[Type[boto3.Session]], optional
            boto3 session instance, by default None
        config : Optional[Type[Config]], optional
            botocore config instance to configure client, by default None

        Returns
        -------
        Type[DynamoDBServiceResource]
            Instance of a boto3 resource client for Parameters feature (e.g., dynamodb, etc.)
        """
        if client is not None:
            user_agent.register_feature_to_resource(resource=client, feature="parameters")
            return client

        session = session or boto3.Session()
        config = config or Config()
        client = session.resource(service_name=service_name, config=config, endpoint_url=endpoint_url)
        user_agent.register_feature_to_resource(resource=client, feature="parameters")
        return client

Ancestors

• abc.ABC

Subclasses

• AppConfigProvider
• DynamoDBProvider
• SecretsProvider
• SSMProvider

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: Optional[int] = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options) ‑> Union[str, dict, bytes, ForwardRef(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.

def get_multiple(self, path: str, max_age: Optional[int] = None, transform: TransformOptions = 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.

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 DynamoDBProvider (table_name: str, key_attr: str = 'id', sort_attr: str = 'sk', value_attr: str = 'value', endpoint_url: Optional[str] = None, config: Optional[botocore.config.Config] = None, boto3_session: Optional[boto3.session.Session] = None, boto3_client: Optional[ForwardRef('DynamoDBServiceResource')] = 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

boto3_session : boto3.session.Session, optional

Boto3 session to create a boto3_client from

boto3_client : DynamoDBServiceResource, optional

Boto3 DynamoDB Resource Client to use; boto3_session will be ignored if both are provided

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
    boto3_session : boto3.session.Session, optional
            Boto3 session to create a boto3_client from
    boto3_client: DynamoDBServiceResource, optional
            Boto3 DynamoDB Resource Client to use; boto3_session will be ignored if both are provided

    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
    """

    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,
        boto3_session: Optional[boto3.session.Session] = None,
        boto3_client: Optional["DynamoDBServiceResource"] = None,
    ):
        """
        Initialize the DynamoDB client
        """
        self.table: "Table" = self._build_boto3_resource_client(
            service_name="dynamodb",
            client=boto3_client,
            session=boto3_session,
            config=config,
            endpoint_url=endpoint_url,
        ).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}

        # maintenance: look for better ways to correctly type DynamoDB multiple return types
        # without a breaking change within ABC return type
        return self.table.get_item(**sdk_options)["Item"][self.value_attr]  # type: ignore[return-value]

    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", []))

        # maintenance: look for better ways to correctly type DynamoDB multiple return types
        # without a breaking change within ABC return type
        return {item[self.sort_attr]: item[self.value_attr] for item in items}

Ancestors

• BaseProvider
• abc.ABC

Inherited members

• BaseProvider:
  • get
  • get_multiple
  • set

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: Optional[Config] = None, boto3_session: Optional[boto3.session.Session] = None, boto3_client: "Optional['SSMClient']" = None)

AWS Systems Manager Parameter Store Provider

Parameters

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

boto3_session : boto3.session.Session, optional

Boto3 session to create a boto3_client from

boto3_client : SSMClient, optional

Boto3 SSM Client to use, boto3_session will be ignored if both are provided

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
    boto3_session : boto3.session.Session, optional
            Boto3 session to create a boto3_client from
    boto3_client: SSMClient, optional
            Boto3 SSM Client to use, boto3_session will be ignored if both are provided

    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
    _MAX_GET_PARAMETERS_ITEM = 10
    _ERRORS_KEY = "_errors"

    def __init__(
        self,
        config: Optional[Config] = None,
        boto3_session: Optional[boto3.session.Session] = None,
        boto3_client: Optional["SSMClient"] = None,
    ):
        """
        Initialize the SSM Parameter Store client
        """

        super().__init__()

        self.client: "SSMClient" = self._build_boto3_client(
            service_name="ssm",
            client=boto3_client,
            session=boto3_session,
            config=config,
        )

    # We break Liskov substitution principle due to differences in signatures of this method and superclass get method
    # We ignore mypy error, as changes to the signature here or in a superclass is a breaking change to users
    def get(  # type: ignore[override]
        self,
        name: str,
        max_age: Optional[int] = None,
        transform: TransformOptions = None,
        decrypt: Optional[bool] = None,
        force_fetch: bool = False,
        **sdk_options,
    ) -> Optional[Union[str, dict, bytes]]:
        """
        Retrieve a parameter value or return the cached value

        Parameters
        ----------
        name: str
            Parameter name
        max_age: int, optional
            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.
        """

        # 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 decrypt is not set, resolve it from the environment variable, defaulting to False
        decrypt = resolve_truthy_env_var_choice(
            env=os.getenv(constants.PARAMETERS_SSM_DECRYPT_ENV, "false"),
            choice=decrypt,
        )

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

    @overload
    def set(
        self,
        name: str,
        value: list[str],
        *,
        overwrite: bool = False,
        description: str = "",
        parameter_type: Literal["StringList"] = "StringList",
        tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard",
        kms_key_id: str | None = "None",
        **sdk_options,
    ): ...

    @overload
    def set(
        self,
        name: str,
        value: str,
        *,
        overwrite: bool = False,
        description: str = "",
        parameter_type: Literal["SecureString"] = "SecureString",
        tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard",
        kms_key_id: str,
        **sdk_options,
    ): ...

    @overload
    def set(
        self,
        name: str,
        value: str,
        *,
        overwrite: bool = False,
        description: str = "",
        parameter_type: Literal["String"] = "String",
        tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard",
        kms_key_id: str | None = None,
        **sdk_options,
    ): ...

    def set(
        self,
        name: str,
        value: str | list[str],
        *,
        overwrite: bool = False,
        description: str = "",
        parameter_type: SSM_PARAMETER_TYPES = "String",
        tier: SSM_PARAMETER_TIER = "Standard",
        kms_key_id: str | None = None,
        **sdk_options,
    ) -> PutParameterResponse:
        """
        Sets a parameter in AWS Systems Manager Parameter Store.

        Parameters
        ----------
        name: str
            The fully qualified name includes the complete hierarchy of the parameter name and name.
        value: str
            The parameter value
        overwrite: bool, optional
            If the parameter value should be overwritten, False by default
        description: str, optional
            The description of the parameter
        parameter_type: str, optional
            Type of the parameter.  Allowed values are String, StringList, and SecureString
        tier: str, optional
            The parameter tier to use. Allowed values are Standard, Advanced, and Intelligent-Tiering
        kms_key_id: str, optional
            The KMS key id to use to encrypt the parameter
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Parameter Store get_parameter API call

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

        URLs:
        -------
            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html

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

            >>> from aws_lambda_powertools.utilities import parameters
            >>>
            >>> response = parameters.set_parameter(name="/my/example/parameter", value="More Powertools")
            >>>
            >>> print(response)
            123

        Returns
        -------
        PutParameterResponse
            The dict returned by boto3.
        """
        opts = {
            "Name": name,
            "Value": value,
            "Overwrite": overwrite,
            "Type": parameter_type,
            "Tier": tier,
            "Description": description,
            **sdk_options,
        }

        if kms_key_id:
            opts["KeyId"] = kms_key_id

        try:
            return self.client.put_parameter(**opts)
        except Exception as exc:
            raise SetParameterError(f"Error setting parameter - {str(exc)}") from exc

    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: Optional[bool] = None,
        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
        """

        # If decrypt is not set, resolve it from the environment variable, defaulting to False
        decrypt = resolve_truthy_env_var_choice(
            env=os.getenv(constants.PARAMETERS_SSM_DECRYPT_ENV, "false"),
            choice=decrypt,
        )

        # 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 contain 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

    # NOTE: When bandwidth permits, allocate a week to refactor to lower cognitive load
    def get_parameters_by_name(
        self,
        parameters: Dict[str, Dict],
        transform: TransformOptions = None,
        decrypt: Optional[bool] = None,
        max_age: Optional[int] = None,
        raise_on_error: bool = True,
    ) -> Dict[str, str] | Dict[str, bytes] | Dict[str, dict]:
        """
        Retrieve multiple parameter values by name from SSM or cache.

        Raise_on_error decides on error handling strategy:

        - A) Default to fail-fast. Raises GetParameterError upon any error
        - B) Gracefully aggregate all parameters that failed under "_errors" key

        It transparently uses GetParameter and/or GetParameters depending on decryption requirements.

                                    ┌────────────────────────┐
                                ┌───▶  Decrypt entire batch  │─────┐
                                │   └────────────────────────┘     │     ┌────────────────────┐
                                │                                  ├─────▶ GetParameters API  │
        ┌──────────────────┐    │   ┌────────────────────────┐     │     └────────────────────┘
        │   Split batch    │─── ┼──▶│ No decryption required │─────┘
        └──────────────────┘    │   └────────────────────────┘
                                │                                        ┌────────────────────┐
                                │   ┌────────────────────────┐           │  GetParameter API  │
                                └──▶│Decrypt some but not all│───────────▶────────────────────┤
                                    └────────────────────────┘           │ GetParameters API  │
                                                                         └────────────────────┘

        Parameters
        ----------
        parameters: List[Dict[str, Dict]]
            List of parameter names, and any optional overrides
        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
        max_age: int, optional
            Maximum age of the cached value
        raise_on_error: bool
            Whether to fail-fast or fail gracefully by including "_errors" key in the response, by default True

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

            When "_errors" reserved key is in parameters to be fetched from SSM.
        """

        # 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 decrypt is not set, resolve it from the environment variable, defaulting to False
        decrypt = resolve_truthy_env_var_choice(
            env=os.getenv(constants.PARAMETERS_SSM_DECRYPT_ENV, "false"),
            choice=decrypt,
        )

        # Init potential batch/decrypt batch responses and errors
        batch_ret: Dict[str, Any] = {}
        decrypt_ret: Dict[str, Any] = {}
        batch_err: List[str] = []
        decrypt_err: List[str] = []
        response: Dict[str, Any] = {}

        # NOTE: We fail early to avoid unintended graceful errors being replaced with their '_errors' param values
        self._raise_if_errors_key_is_present(parameters, self._ERRORS_KEY, raise_on_error)

        batch_params, decrypt_params = self._split_batch_and_decrypt_parameters(parameters, transform, max_age, decrypt)

        # NOTE: We need to find out whether all parameters must be decrypted or not to know which API to use
        ## Logic:
        ##
        ## GetParameters API -> When decrypt is used for all parameters in the the batch
        ## GetParameter  API -> When decrypt is used for one or more in the batch

        if len(decrypt_params) != len(parameters):
            decrypt_ret, decrypt_err = self._get_parameters_by_name_with_decrypt_option(decrypt_params, raise_on_error)
            batch_ret, batch_err = self._get_parameters_batch_by_name(batch_params, raise_on_error, decrypt=False)
        else:
            batch_ret, batch_err = self._get_parameters_batch_by_name(decrypt_params, raise_on_error, decrypt=True)

        # Fail-fast disabled, let's aggregate errors under "_errors" key so they can handle gracefully
        if not raise_on_error:
            response[self._ERRORS_KEY] = [*decrypt_err, *batch_err]

        return {**response, **batch_ret, **decrypt_ret}

    def _get_parameters_by_name_with_decrypt_option(
        self,
        batch: Dict[str, Dict],
        raise_on_error: bool,
    ) -> Tuple[Dict, List]:
        response: Dict[str, Any] = {}
        errors: List[str] = []

        # Decided for single-thread as it outperforms in 128M and 1G + reduce timeout risk
        # see: https://github.com/aws-powertools/powertools-lambda-python/issues/1040#issuecomment-1299954613
        for parameter, options in batch.items():
            try:
                response[parameter] = self.get(parameter, options["max_age"], options["transform"], options["decrypt"])
            except GetParameterError:
                if raise_on_error:
                    raise
                errors.append(parameter)
                continue

        return response, errors

    def _get_parameters_batch_by_name(
        self,
        batch: Dict[str, Dict],
        raise_on_error: bool = True,
        decrypt: bool = False,
    ) -> Tuple[Dict, List]:
        """Slice batch and fetch parameters using GetParameters by max permitted"""
        errors: List[str] = []

        # Fetch each possible batch param from cache and return if entire batch is cached
        cached_params = self._get_parameters_by_name_from_cache(batch)
        if len(cached_params) == len(batch):
            return cached_params, errors

        # Slice batch by max permitted GetParameters call
        batch_ret, errors = self._get_parameters_by_name_in_chunks(batch, cached_params, raise_on_error, decrypt)

        return {**cached_params, **batch_ret}, errors

    def _get_parameters_by_name_from_cache(self, batch: Dict[str, Dict]) -> Dict[str, Any]:
        """Fetch each parameter from batch that hasn't been expired"""
        cache = {}
        for name, options in batch.items():
            cache_key = (name, options["transform"])
            if self.has_not_expired_in_cache(cache_key):
                cache[name] = self.store[cache_key].value

        return cache

    def _get_parameters_by_name_in_chunks(
        self,
        batch: Dict[str, Dict],
        cache: Dict[str, Any],
        raise_on_error: bool,
        decrypt: bool = False,
    ) -> Tuple[Dict, List]:
        """Take out differences from cache and batch, slice it and fetch from SSM"""
        response: Dict[str, Any] = {}
        errors: List[str] = []

        diff = {key: value for key, value in batch.items() if key not in cache}

        for chunk in slice_dictionary(data=diff, chunk_size=self._MAX_GET_PARAMETERS_ITEM):
            response, possible_errors = self._get_parameters_by_name(
                parameters=chunk,
                raise_on_error=raise_on_error,
                decrypt=decrypt,
            )
            response.update(response)
            errors.extend(possible_errors)

        return response, errors

    def _get_parameters_by_name(
        self,
        parameters: Dict[str, Dict],
        raise_on_error: bool = True,
        decrypt: bool = False,
    ) -> Tuple[Dict[str, Any], List[str]]:
        """Use SSM GetParameters to fetch parameters, hydrate cache, and handle partial failure

        Parameters
        ----------
        parameters : Dict[str, Dict]
            Parameters to fetch
        raise_on_error : bool, optional
            Whether to fail-fast or fail gracefully by including "_errors" key in the response, by default True

        Returns
        -------
        Dict[str, Any]
            Retrieved parameters as key names and their values

        Raises
        ------
        GetParameterError
            When one or more parameters failed on fetching, and raise_on_error is enabled
        """
        ret: Dict[str, Any] = {}
        batch_errors: List[str] = []
        parameter_names = list(parameters.keys())

        # All params in the batch must be decrypted
        # we return early if we hit an unrecoverable exception like InvalidKeyId/InternalServerError
        # everything else should technically be recoverable as GetParameters is non-atomic
        try:
            if decrypt:
                response = self.client.get_parameters(Names=parameter_names, WithDecryption=True)
            else:
                response = self.client.get_parameters(Names=parameter_names)
        except (self.client.exceptions.InvalidKeyId, self.client.exceptions.InternalServerError):
            return ret, parameter_names

        batch_errors = self._handle_any_invalid_get_parameter_errors(response, raise_on_error)
        transformed_params = self._transform_and_cache_get_parameters_response(response, parameters, raise_on_error)

        return transformed_params, batch_errors

    def _transform_and_cache_get_parameters_response(
        self,
        api_response: GetParametersResultTypeDef,
        parameters: Dict[str, Any],
        raise_on_error: bool = True,
    ) -> Dict[str, Any]:
        response: Dict[str, Any] = {}

        for parameter in api_response["Parameters"]:
            name = parameter["Name"]
            value = parameter["Value"]
            options = parameters[name]
            transform = options.get("transform")

            # NOTE: If transform is set, we do it before caching to reduce number of operations
            if transform:
                value = transform_value(name, value, transform, raise_on_error)  # type: ignore

            _cache_key = (name, options["transform"])
            self.add_to_cache(key=_cache_key, value=value, max_age=options["max_age"])

            response[name] = value

        return response

    @staticmethod
    def _handle_any_invalid_get_parameter_errors(
        api_response: GetParametersResultTypeDef,
        raise_on_error: bool = True,
    ) -> List[str]:
        """GetParameters is non-atomic. Failures don't always reflect in exceptions so we need to collect."""
        failed_parameters = api_response["InvalidParameters"]
        if failed_parameters:
            if raise_on_error:
                raise GetParameterError(f"Failed to fetch parameters: {failed_parameters}")

            return failed_parameters

        return []

    @staticmethod
    def _split_batch_and_decrypt_parameters(
        parameters: Dict[str, Dict],
        transform: TransformOptions,
        max_age: int,
        decrypt: bool,
    ) -> Tuple[Dict[str, Dict], Dict[str, Dict]]:
        """Split parameters that can be fetched by GetParameters vs GetParameter

        Parameters
        ----------
        parameters : Dict[str, Dict]
            Parameters containing names as key and optional config override as value
        transform : TransformOptions
            Transform configuration
        max_age : int
            How long to cache a parameter for
        decrypt : bool
            Whether to use KMS to decrypt a parameter

        Returns
        -------
        Tuple[Dict[str, Dict], Dict[str, Dict]]
            GetParameters and GetParameter parameters dict along with their overrides/globals merged
        """
        batch_parameters: Dict[str, Dict] = {}
        decrypt_parameters: Dict[str, Any] = {}

        for parameter, options in parameters.items():
            # NOTE: TypeDict later
            _overrides = options or {}
            _overrides["transform"] = _overrides.get("transform") or transform

            # These values can be falsy (False, 0)
            if "decrypt" not in _overrides:
                _overrides["decrypt"] = decrypt

            if "max_age" not in _overrides:
                _overrides["max_age"] = max_age

            # NOTE: Split parameters who have decrypt OR have it global
            if _overrides["decrypt"]:
                decrypt_parameters[parameter] = _overrides
            else:
                batch_parameters[parameter] = _overrides

        return batch_parameters, decrypt_parameters

    @staticmethod
    def _raise_if_errors_key_is_present(parameters: Dict, reserved_parameter: str, raise_on_error: bool):
        """Raise GetParameterError if fail-fast is disabled and '_errors' key is in parameters batch"""
        if not raise_on_error and reserved_parameter in parameters:
            raise GetParameterError(
                f"You cannot fetch a parameter named '{reserved_parameter}' in graceful error mode.",
            )

Ancestors

• BaseProvider
• abc.ABC

Class variables

var client : Any

Methods

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

Retrieve a parameter value or return the cached value

Parameters

name : str

Parameter name

max_age : int, optional

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.

def get_parameters_by_name(self, parameters: Dict[str, Dict], transform: TransformOptions = None, decrypt: Optional[bool] = None, max_age: Optional[int] = None, raise_on_error: bool = True) ‑> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]

Retrieve multiple parameter values by name from SSM or cache.

Raise_on_error decides on error handling strategy:

• A) Default to fail-fast. Raises GetParameterError upon any error
• B) Gracefully aggregate all parameters that failed under "_errors" key

It transparently uses GetParameter and/or GetParameters depending on decryption requirements.

                        ┌────────────────────────┐
                    ┌───▶  Decrypt entire batch  │─────┐
                    │   └────────────────────────┘     │     ┌────────────────────┐
                    │                                  ├─────▶ GetParameters API  │

┌──────────────────┐ │ ┌────────────────────────┐ │ └────────────────────┘ │ Split batch │─── ┼──▶│ No decryption required │─────┘ └──────────────────┘ │ └────────────────────────┘ │ ┌────────────────────┐ │ ┌────────────────────────┐ │ GetParameter API │ └──▶│Decrypt some but not all│───────────▶────────────────────┤ └────────────────────────┘ │ GetParameters API │ └────────────────────┘

Parameters

parameters : List[Dict[str, Dict]]

List of parameter names, and any optional overrides

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

max_age : int, optional

Maximum age of the cached value

raise_on_error : bool

Whether to fail-fast or fail gracefully by including "_errors" key in the response, by default True

Raises

GetParameterError

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

When "_errors" reserved key is in parameters to be fetched from SSM.

def set(self, name: str, value: str | list[str], *, overwrite: bool = False, description: str = '', parameter_type: SSM_PARAMETER_TYPES = 'String', tier: SSM_PARAMETER_TIER = 'Standard', kms_key_id: str | None = None, **sdk_options) ‑> PutParameterResponse

Sets a parameter in AWS Systems Manager Parameter Store.

Parameters

name : str

The fully qualified name includes the complete hierarchy of the parameter name and name.

value : str

The parameter value

overwrite : bool, optional

If the parameter value should be overwritten, False by default

description : str, optional

The description of the parameter

parameter_type : str, optional

Type of the parameter. Allowed values are String, StringList, and SecureString

tier : str, optional

The parameter tier to use. Allowed values are Standard, Advanced, and Intelligent-Tiering

kms_key_id : str, optional

The KMS key id to use to encrypt the parameter

sdk_options : dict, optional

Dictionary of options that will be passed to the Parameter Store get_parameter API call

Raises

SetParameterError

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

URLs:

<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html>

Example

Sets a parameter value from Systems Manager Parameter Store

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> response = parameters.set_parameter(name="/my/example/parameter", value="More Powertools")
>>>
>>> print(response)
123

Returns

PutParameterResponse

The dict returned by boto3.

Inherited members

• BaseProvider:
  • get_multiple

class SecretsProvider (config: Optional[Config] = None, boto3_session: Optional[boto3.session.Session] = None, boto3_client: "Optional['SecretsManagerClient']" = None)

AWS Secrets Manager Parameter Provider

Parameters

config : botocore.config.Config, optional

Botocore configuration to pass during client initialization

boto3_session : boto3.session.Session, optional

Boto3 session to create a boto3_client from

boto3_client : SecretsManagerClient, optional

Boto3 SecretsManager Client to use, boto3_session will be ignored if both are provided

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
    boto3_session : boto3.session.Session, optional
            Boto3 session to create a boto3_client from
    boto3_client: SecretsManagerClient, optional
            Boto3 SecretsManager Client to use, boto3_session will be ignored if both are provided

    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,
        boto3_session: Optional[boto3.session.Session] = None,
        boto3_client: Optional["SecretsManagerClient"] = None,
    ):
        """
        Initialize the Secrets Manager client
        """

        super().__init__()

        self.client: "SecretsManagerClient" = self._build_boto3_client(
            service_name="secretsmanager",
            client=boto3_client,
            session=boto3_session,
            config=config,
        )

    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

        secret_value = self.client.get_secret_value(**sdk_options)

        if "SecretString" in secret_value:
            return secret_value["SecretString"]

        return secret_value["SecretBinary"]

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

    def _create_secret(self, name: str, **sdk_options):
        """
        Create a secret with the given name.

        Parameters:
        ----------
        name: str
            The name of the secret.
        **sdk_options:
            Additional options to be passed to the create_secret method.

        Raises:
            SetSecretError: If there is an error setting the secret.
        """
        try:
            sdk_options["Name"] = name
            return self.client.create_secret(**sdk_options)
        except Exception as exc:
            raise SetSecretError(f"Error setting secret - {str(exc)}") from exc

    def _update_secret(self, name: str, **sdk_options):
        """
        Update a secret with the given name.

        Parameters:
        ----------
        name: str
            The name of the secret.
        **sdk_options:
            Additional options to be passed to the create_secret method.
        """
        sdk_options["SecretId"] = name
        return self.client.put_secret_value(**sdk_options)

    def set(
        self,
        name: str,
        value: Union[str, dict, bytes],
        *,  # force keyword arguments
        client_request_token: Optional[str] = None,
        **sdk_options,
    ) -> SetSecretResponse:
        """
        Modify the details of a secret or create a new secret if it doesn't already exist.

        We aim to minimize API calls by assuming that the secret already exists and needs updating.
        If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding:


                          ┌────────────────────────┐      ┌─────────────────┐
                ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐
                │         └────────────────────────┘      └─────────────────┘     │
                │                                                                 │
                │                                                                 │
                │                                                                 ▼
        ┌─────────────────┐                                              ┌─────────────────────┐
        │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │
        └─────────────────┘                                              └─────────────────────┘

        Parameters
        ----------
        name: str
            The ARN or name of the secret to add a new version to or create a new one.
        value: str, dict or bytes
            Specifies text data that you want to encrypt and store in this new version of the secret.
        client_request_token: str, optional
            This value helps ensure idempotency. It's recommended that you generate
            a UUID-type value to ensure uniqueness within the specified secret.
            This value becomes the VersionId of the new version. This field is
            auto-populated if not provided, but no idempotency will be enforced this way.
        sdk_options: dict, optional
            Dictionary of options that will be passed to the Secrets Manager update_secret API call

        Raises
        ------
        SetSecretError
            When attempting to update or create a secret fails.

        Returns:
        -------
        SetSecretResponse:
            The dict returned by boto3.

        Example
        -------
        **Sets a secret***

            >>> from aws_lambda_powertools.utilities import parameters
            >>>
            >>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd")

        **Sets a secret and includes an client_request_token**

            >>> from aws_lambda_powertools.utilities import parameters
            >>> import uuid
            >>>
            >>> parameters.set_secret(
                    name="my-secret",
                    value='{"password": "supers3cr3tllam@passw0rd"}',
                    client_request_token=str(uuid.uuid4())
                )

        URLs:
        -------
            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html
            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html
        """

        if isinstance(value, dict):
            value = json.dumps(value, cls=Encoder)

        if isinstance(value, bytes):
            sdk_options["SecretBinary"] = value
        else:
            sdk_options["SecretString"] = value

        if client_request_token:
            sdk_options["ClientRequestToken"] = client_request_token

        try:
            logger.debug(f"Attempting to update secret {name}")
            return self._update_secret(name=name, **sdk_options)
        except self.client.exceptions.ResourceNotFoundException:
            logger.debug(f"Secret {name} doesn't exist, creating a new one")
            return self._create_secret(name=name, **sdk_options)
        except Exception as exc:
            raise SetSecretError(f"Error setting secret - {str(exc)}") from exc

Ancestors

• BaseProvider
• abc.ABC

Class variables

var client : Any

Methods

def set(self, name: str, value: Union[str, dict, bytes], *, client_request_token: Optional[str] = None, **sdk_options) ‑> SetSecretResponse

Modify the details of a secret or create a new secret if it doesn't already exist.

We aim to minimize API calls by assuming that the secret already exists and needs updating. If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding:

              ┌────────────────────────┐      ┌─────────────────┐
    ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐
    │         └────────────────────────┘      └─────────────────┘     │
    │                                                                 │
    │                                                                 │
    │                                                                 ▼

┌─────────────────┐ ┌─────────────────────┐ │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │ └─────────────────┘ └─────────────────────┘

Parameters

name : str

The ARN or name of the secret to add a new version to or create a new one.

value : str, dict or bytes

Specifies text data that you want to encrypt and store in this new version of the secret.

client_request_token : str, optional

This value helps ensure idempotency. It's recommended that you generate a UUID-type value to ensure uniqueness within the specified secret. This value becomes the VersionId of the new version. This field is auto-populated if not provided, but no idempotency will be enforced this way.

sdk_options : dict, optional

Dictionary of options that will be passed to the Secrets Manager update_secret API call

Raises

SetSecretError

When attempting to update or create a secret fails.

Returns:

Setsecretresponse

The dict returned by boto3.

Example

Sets a secret*

>>> from aws_lambda_powertools.utilities import parameters
>>>
>>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd")

Sets a secret and includes an client_request_token

>>> from aws_lambda_powertools.utilities import parameters
>>> import uuid
>>>
>>> parameters.set_secret(
        name="my-secret",
        value='{"password": "supers3cr3tllam@passw0rd"}',
        client_request_token=str(uuid.uuid4())
    )

URLs:

<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html>
<https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html>

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