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.constants
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: str | None = None, transform: TransformOptions = None, force_fetch: bool = False, max_age: int | None = None, **sdk_options)
-
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: bool | None = None, force_fetch: bool = False, max_age: int | None = None, **sdk_options)
-
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: bool | None = None, force_fetch: bool = False, max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options)
-
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: bool | None = None, max_age: int | None = None, raise_on_error: bool = True)
-
Retrieve multiple parameter values by name from AWS Systems Manager (SSM) Parameter Store
Parameters
parameters
:dict[str, Any]
- 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: int | None = None, **sdk_options)
-
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)
-
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
PutParameterResultTypeDef
- The dict returned by boto3.
def set_secret(name: str, value: str | bytes, *, client_request_token: str | None = None, **sdk_options)
-
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
orbytes
- 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: str | None = None, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: AppConfigDataClient | None = 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 """ def __init__( self, environment: str, application: str | None = None, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: AppConfigDataClient | None = None, ): """ Initialize the App Config client """ super().__init__() if config: warnings.warn( message="The 'config' parameter is deprecated in V3 and will be removed in V4. " "Please use 'boto_config' instead.", category=PowertoolsDeprecationWarning, stacklevel=2, ) if boto3_client is None: boto3_session = boto3_session or boto3.session.Session() boto3_client = boto3_session.client("appconfigdata", config=boto_config or config) self.client = boto3_client 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, bytes] = {} super().__init__(client=self.client) def _get(self, name: str, **sdk_options) -> bytes: """ 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
Inherited members
class BaseProvider (*, client=None, resource=None)
-
Abstract Base Class for Parameter providers
Initialize the base provider
Expand source code
class BaseProvider(ABC): """ Abstract Base Class for Parameter providers """ store: dict[tuple, ExpirableValue] def __init__(self, *, client=None, resource=None): """ Initialize the base provider """ if client is not None: user_agent.register_feature_to_client(client=client, feature="parameters") if resource is not None: user_agent.register_feature_to_resource(resource=resource, feature="parameters") self.store: dict[tuple, ExpirableValue] = {} def has_not_expired_in_cache(self, key: tuple) -> bool: return key in self.store and self.store[key].ttl >= datetime.now() def get( self, name: str, max_age: int | None = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options, ) -> str | bytes | dict | None: """ Retrieve a parameter value or return the cached value Parameters ---------- name: str Parameter name max_age: int Maximum age of the cached value transform: str Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values. force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Arguments that will be passed directly to the underlying API call Raises ------ GetParameterError When the parameter provider fails to retrieve a parameter value for a given name. TransformParameterError When the parameter provider fails to transform a parameter value. """ # If there are multiple calls to the same parameter but in a different # transform, they will be stored multiple times. This allows us to # optimize by transforming the data only once per retrieval, thus there # is no need to transform cached values multiple times. However, this # means that we need to make multiple calls to the underlying parameter # store if we need to return it in different transforms. Since the number # of supported transform is small and the probability that a given # parameter will always be used in a specific transform, this should be # an acceptable tradeoff. value: str | bytes | dict | None = None key = self._build_cache_key(name=name, transform=transform) # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age) if not force_fetch and self.has_not_expired_in_cache(key): return self.fetch_from_cache(key) try: value = self._get(name, **sdk_options) # Encapsulate all errors into a generic GetParameterError except Exception as exc: raise GetParameterError(str(exc)) if transform: value = transform_value(key=name, value=value, transform=transform, raise_on_transform_error=True) # NOTE: don't cache None, as they might've been failed transforms and may be corrected if value is not None: self.add_to_cache(key=key, value=value, max_age=max_age) return value @abstractmethod def _get(self, name: str, **sdk_options) -> str | bytes | dict[str, Any]: """ Retrieve parameter value from the underlying parameter store """ raise NotImplementedError() def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs): """ Set parameter value from the underlying parameter store """ raise NotImplementedError() def get_multiple( self, path: str, max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, force_fetch: bool = False, **sdk_options, ) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameters based on a path prefix Parameters ---------- path: str Parameter path used to retrieve multiple parameters max_age: int, optional Maximum age of the cached value transform: str, optional Optional transformation of the parameter value. Supported values are "json" for JSON strings, "binary" for base 64 encoded values or "auto" which looks at the attribute key to determine the type. raise_on_transform_error: bool, optional Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Arguments that will be passed directly to the underlying API call Raises ------ GetParameterError When the parameter provider fails to retrieve parameter values for a given path. TransformParameterError When the parameter provider fails to transform a parameter value. """ key = self._build_cache_key(name=path, transform=transform, is_nested=True) # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age) if not force_fetch and self.has_not_expired_in_cache(key): return self.fetch_from_cache(key) try: values = self._get_multiple(path, **sdk_options) # Encapsulate all errors into a generic GetParameterError except Exception as exc: raise GetParameterError(str(exc)) if transform: values.update(transform_value(values, transform, raise_on_transform_error)) self.add_to_cache(key=key, value=values, max_age=max_age) return values @abstractmethod def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieve multiple parameter values from the underlying parameter store """ raise NotImplementedError() def clear_cache(self): self.store.clear() def fetch_from_cache(self, key: tuple): return self.store[key].value if key in self.store else {} def add_to_cache(self, key: tuple, value: Any, max_age: int): if max_age <= 0: return self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age)) def _build_cache_key( self, name: str, transform: TransformOptions = None, is_nested: bool = False, ): """Creates cache key for parameters Parameters ---------- name : str Name of parameter, secret or config transform : TransformOptions, optional Transform method used, by default None is_nested : bool, optional Whether it's a single parameter or multiple nested parameters, by default False Returns ------- tuple[str, TransformOptions, bool] Cache key """ return (name, transform, is_nested)
Ancestors
- abc.ABC
Subclasses
Class variables
var store : dict[tuple, ExpirableValue]
Methods
def add_to_cache(self, key: tuple, value: Any, max_age: int)
def clear_cache(self)
def fetch_from_cache(self, key: tuple)
def get(self, name: str, max_age: int | None = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options)
-
Retrieve a parameter value or return the cached value
Parameters
name
:str
- Parameter name
max_age
:int
- Maximum age of the cached value
transform
:str
- Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values.
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
sdk_options
:dict
, optional- Arguments that will be passed directly to the underlying API call
Raises
GetParameterError
- When the parameter provider fails to retrieve a parameter value for a given name.
TransformParameterError
- When the parameter provider fails to transform a parameter value.
def get_multiple(self, path: str, max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, force_fetch: bool = False, **sdk_options)
-
Retrieve multiple parameters based on a path prefix
Parameters
path
:str
- Parameter path used to retrieve multiple parameters
max_age
:int
, optional- Maximum age of the cached value
transform
:str
, optional- Optional transformation of the parameter value. Supported values are "json" for JSON strings, "binary" for base 64 encoded values or "auto" which looks at the attribute key to determine the type.
raise_on_transform_error
:bool
, optional- Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
sdk_options
:dict
, optional- Arguments that will be passed directly to the underlying API call
Raises
GetParameterError
- When the parameter provider fails to retrieve parameter values for a given path.
TransformParameterError
- When the parameter provider fails to transform a parameter value.
def has_not_expired_in_cache(self, key: tuple) ‑> bool
def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs)
-
Set parameter value from the underlying parameter store
class DynamoDBProvider (table_name: str, key_attr: str = 'id', sort_attr: str = 'sk', value_attr: str = 'value', endpoint_url: str | None = None, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: DynamoDBServiceResource | None = 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 thevalue
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: str | None = None, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: DynamoDBServiceResource | None = None, ): """ Initialize the DynamoDB client """ if config: warnings.warn( message="The 'config' parameter is deprecated in V3 and will be removed in V4. " "Please use 'boto_config' instead.", category=PowertoolsDeprecationWarning, stacklevel=2, ) if boto3_client is None: boto3_session = boto3_session or boto3.session.Session() boto3_client = boto3_session.resource("dynamodb", config=boto_config or config, endpoint_url=endpoint_url) self.table = boto3_client.Table(table_name) self.key_attr = key_attr self.sort_attr = sort_attr self.value_attr = value_attr super().__init__(resource=boto3_client) 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
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: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: SSMClient | None = 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 """ _MAX_GET_PARAMETERS_ITEM = 10 _ERRORS_KEY = "_errors" def __init__( self, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: SSMClient | None = None, ): """ Initialize the SSM Parameter Store client """ if config: warnings.warn( message="The 'config' parameter is deprecated in V3 and will be removed in V4. " "Please use 'boto_config' instead.", category=PowertoolsDeprecationWarning, stacklevel=2, ) if boto3_client is None: boto3_session = boto3_session or boto3.session.Session() boto3_client = boto3_session.client("ssm", config=boto_config or config) self.client = boto3_client super().__init__(client=self.client) def get_multiple( # type: ignore[override] self, path: str, max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, decrypt: bool | None = None, force_fetch: bool = False, recursive: bool = False, **sdk_options, ) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameters based on a path prefix Parameters ---------- path: str Parameter path used to retrieve multiple parameters max_age: int, optional Maximum age of the cached value transform: str, optional Optional transformation of the parameter value. Supported values are "json" for JSON strings, "binary" for base 64 encoded values or "auto" which looks at the attribute key to determine the type. raise_on_transform_error: bool, optional Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed force_fetch: bool, optional Force update even before a cached item has expired, defaults to False recursive: bool, optional If this should retrieve the parameter values recursively or not 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. """ # 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, ) sdk_options["decrypt"] = decrypt sdk_options["recursive"] = recursive return super().get_multiple(path, max_age, transform, force_fetch, raise_on_transform_error, **sdk_options) # 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: int | None = None, transform: TransformOptions = None, decrypt: bool | None = None, force_fetch: bool = False, **sdk_options, ) -> str | bytes | dict | None: """ Retrieve a parameter value or return the cached value Parameters ---------- name: str Parameter name max_age: int, 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, ) -> PutParameterResultTypeDef: """ 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 ------- PutParameterResultTypeDef 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: bool | None = 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 """ # 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: bool | None = None, max_age: int | None = 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: 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
Methods
def get(self, name: str, max_age: int | None = None, transform: TransformOptions = None, decrypt: bool | None = None, force_fetch: bool = False, **sdk_options)
-
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_multiple(self, path: str, max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, decrypt: bool | None = None, force_fetch: bool = False, recursive: bool = False, **sdk_options)
-
Retrieve multiple parameters based on a path prefix
Parameters
path
:str
- Parameter path used to retrieve multiple parameters
max_age
:int
, optional- Maximum age of the cached value
transform
:str
, optional- Optional transformation of the parameter value. Supported values are "json" for JSON strings, "binary" for base 64 encoded values or "auto" which looks at the attribute key to determine the type.
raise_on_transform_error
:bool
, optional- Raises an exception if any transform fails, otherwise this will return a None value for each transform that failed
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
recursive
:bool
, optional- If this should retrieve the parameter values recursively or not
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 get_parameters_by_name(self, parameters: dict[str, dict], transform: TransformOptions = None, decrypt: bool | None = None, max_age: int | None = None, raise_on_error: bool = True)
-
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
: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)
-
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
PutParameterResultTypeDef
- The dict returned by boto3.
class SecretsProvider (config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: SecretsManagerClient | None = 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 """ def __init__( self, config: Config | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: SecretsManagerClient | None = None, ): """ Initialize the Secrets Manager client """ if config: warnings.warn( message="The 'config' parameter is deprecated in V3 and will be removed in V4. " "Please use 'boto_config' instead.", category=PowertoolsDeprecationWarning, stacklevel=2, ) if boto3_client is None: boto3_session = boto3_session or boto3.session.Session() boto3_client = boto3_session.client("secretsmanager", config=boto_config or config) self.client = boto3_client super().__init__(client=self.client) def _get(self, name: str, **sdk_options) -> str | bytes: """ 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) -> CreateSecretResponseTypeDef: """ 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: str | bytes | dict, *, # force keyword arguments client_request_token: str | None = None, **sdk_options, ) -> CreateSecretResponseTypeDef: """ 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
Methods
def set(self, name: str, value: str | bytes | dict, *, client_request_token: str | None = None, **sdk_options)
-
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
orbytes
- 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
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