Module aws_lambda_powertools.utilities.parameters.ssm
AWS SSM Parameter retrieval and caching utility
Expand source code
"""
AWS SSM Parameter retrieval and caching utility
"""
from typing import Any, Dict, Optional, Union
import boto3
from botocore.config import Config
from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
class SSMProvider(BaseProvider):
"""
AWS Systems Manager Parameter Store Provider
Parameters
----------
config: botocore.config.Config, optional
Botocore configuration to pass during client initialization
Example
-------
**Retrieves a parameter value from Systems Manager Parameter Store**
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> value = ssm_provider.get("/my/parameter")
>>>
>>> print(value)
My parameter value
**Retrieves a parameter value from Systems Manager Parameter Store in another AWS region**
>>> from botocore.config import Config
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>>
>>> config = Config(region_name="us-west-1")
>>> ssm_provider = SSMProvider(config=config)
>>>
>>> value = ssm_provider.get("/my/parameter")
>>>
>>> print(value)
My parameter value
**Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix**
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> values = ssm_provider.get_multiple("/my/path/prefix")
>>>
>>> for key, value in values.items():
... print(key, value)
/my/path/prefix/a Parameter value a
/my/path/prefix/b Parameter value b
/my/path/prefix/c Parameter value c
**Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call**
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider
>>> ssm_provider = SSMProvider()
>>>
>>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10)
>>>
>>> for key, value in values.items():
... print(key, value)
/my/path/prefix/a Parameter value a
/my/path/prefix/b Parameter value b
/my/path/prefix/c Parameter value c
"""
client: Any = None
def __init__(self, config: Optional[Config] = None):
"""
Initialize the SSM Parameter Store client
"""
config = config or Config()
self.client = boto3.client("ssm", config=config)
super().__init__()
def get(
self,
name: str,
max_age: int = DEFAULT_MAX_AGE_SECS,
transform: Optional[str] = None,
decrypt: bool = False,
force_fetch: bool = False,
**sdk_options
) -> Union[str, list, dict, bytes]:
"""
Retrieve a parameter value or return the cached value
Parameters
----------
name: str
Parameter name
max_age: int
Maximum age of the cached value
transform: str
Optional transformation of the parameter value. Supported values
are "json" for JSON strings and "binary" for base 64 encoded
values.
decrypt: bool, optional
If the parameter value should be decrypted
force_fetch: bool, optional
Force update even before a cached item has expired, defaults to False
sdk_options: dict, optional
Arguments that will be passed directly to the underlying API call
Raises
------
GetParameterError
When the parameter provider fails to retrieve a parameter value for
a given name.
TransformParameterError
When the parameter provider fails to transform a parameter value.
"""
# Add to `decrypt` sdk_options to we can have an explicit option for this
sdk_options["decrypt"] = decrypt
return super().get(name, max_age, transform, force_fetch, **sdk_options)
def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str:
"""
Retrieve a parameter value from AWS Systems Manager Parameter Store
Parameters
----------
name: str
Parameter name
decrypt: bool, optional
If the parameter value should be decrypted
sdk_options: dict, optional
Dictionary of options that will be passed to the Parameter Store get_parameter API call
"""
# Explicit arguments will take precedence over keyword arguments
sdk_options["Name"] = name
sdk_options["WithDecryption"] = decrypt
return self.client.get_parameter(**sdk_options)["Parameter"]["Value"]
def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = False, **sdk_options) -> Dict[str, str]:
"""
Retrieve multiple parameter values from AWS Systems Manager Parameter Store
Parameters
----------
path: str
Path to retrieve the parameters
decrypt: bool, optional
If the parameter values should be decrypted
recursive: bool, optional
If this should retrieve the parameter values recursively or not
sdk_options: dict, optional
Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call
"""
# Explicit arguments will take precedence over keyword arguments
sdk_options["Path"] = path
sdk_options["WithDecryption"] = decrypt
sdk_options["Recursive"] = recursive
parameters = {}
for page in self.client.get_paginator("get_parameters_by_path").paginate(**sdk_options):
for parameter in page.get("Parameters", []):
# Standardize the parameter name
# The parameter name returned by SSM will contained the full path.
# However, for readability, we should return only the part after
# the path.
name = parameter["Name"]
if name.startswith(path):
name = name[len(path) :]
name = name.lstrip("/")
parameters[name] = parameter["Value"]
return parameters
def get_parameter(
name: str, transform: Optional[str] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options
) -> Union[str, list, dict, bytes]:
"""
Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store
Parameters
----------
name: str
Name of the parameter
transform: str, optional
Transforms the content from a JSON object ('json') or base64 binary string ('binary')
decrypt: bool, optional
If the parameter values should be decrypted
force_fetch: bool, optional
Force update even before a cached item has expired, defaults to False
sdk_options: dict, optional
Dictionary of options that will be passed to the Parameter Store get_parameter API call
Raises
------
GetParameterError
When the parameter provider fails to retrieve a parameter value for
a given name.
TransformParameterError
When the parameter provider fails to transform a parameter value.
Example
-------
**Retrieves a parameter value from Systems Manager Parameter Store**
>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> value = get_parameter("/my/parameter")
>>>
>>> print(value)
My parameter value
**Retrieves a parameter value and decodes it using a Base64 decoder**
>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> value = get_parameter("/my/parameter", transform='binary')
>>>
>>> print(value)
My parameter value
"""
# Only create the provider if this function is called at least once
if "ssm" not in DEFAULT_PROVIDERS:
DEFAULT_PROVIDERS["ssm"] = SSMProvider()
# Add to `decrypt` sdk_options to we can have an explicit option for this
sdk_options["decrypt"] = decrypt
return DEFAULT_PROVIDERS["ssm"].get(name, transform=transform, force_fetch=force_fetch, **sdk_options)
def get_parameters(
path: str,
transform: Optional[str] = None,
recursive: bool = True,
decrypt: bool = False,
force_fetch: bool = False,
**sdk_options
) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
"""
Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store
Parameters
----------
path: str
Path to retrieve the parameters
transform: str, optional
Transforms the content from a JSON object ('json') or base64 binary string ('binary')
recursive: bool, optional
If this should retrieve the parameter values recursively or not, defaults to True
decrypt: bool, optional
If the parameter values should be decrypted
force_fetch: bool, optional
Force update even before a cached item has expired, defaults to False
sdk_options: dict, optional
Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call
Raises
------
GetParameterError
When the parameter provider fails to retrieve parameter values for
a given path.
TransformParameterError
When the parameter provider fails to transform a parameter value.
Example
-------
**Retrieves parameter values from Systems Manager Parameter Store**
>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> values = get_parameters("/my/path/prefix")
>>>
>>> for key, value in values.items():
... print(key, value)
/my/path/prefix/a Parameter value a
/my/path/prefix/b Parameter value b
/my/path/prefix/c Parameter value c
**Retrieves parameter values and decodes them using a Base64 decoder**
>>> from aws_lambda_powertools.utilities.parameters import get_parameter
>>>
>>> values = get_parameters("/my/path/prefix", transform='binary')
"""
# Only create the provider if this function is called at least once
if "ssm" not in DEFAULT_PROVIDERS:
DEFAULT_PROVIDERS["ssm"] = SSMProvider()
sdk_options["recursive"] = recursive
sdk_options["decrypt"] = decrypt
return DEFAULT_PROVIDERS["ssm"].get_multiple(path, transform=transform, force_fetch=force_fetch, **sdk_options)
Functions
def get_parameter(name: str, transform: Union[str, NoneType] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]
-
Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store
Parameters
name
:str
- Name of the parameter
transform
:str
, optional- Transforms the content from a JSON object ('json') or base64 binary string ('binary')
decrypt
:bool
, optional- If the parameter values should be decrypted
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
sdk_options
:dict
, optional- Dictionary of options that will be passed to the Parameter Store get_parameter API call
Raises
GetParameterError
- When the parameter provider fails to retrieve a parameter value for a given name.
TransformParameterError
- When the parameter provider fails to transform a parameter value.
Example
Retrieves a parameter value from Systems Manager Parameter Store
>>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> value = get_parameter("/my/parameter") >>> >>> print(value) My parameter value
Retrieves a parameter value and decodes it using a Base64 decoder
>>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> value = get_parameter("/my/parameter", transform='binary') >>> >>> print(value) My parameter value
Expand source code
def get_parameter( name: str, transform: Optional[str] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options ) -> Union[str, list, dict, bytes]: """ Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store Parameters ---------- name: str Name of the parameter transform: str, optional Transforms the content from a JSON object ('json') or base64 binary string ('binary') decrypt: bool, optional If the parameter values should be decrypted force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Dictionary of options that will be passed to the Parameter Store get_parameter API call Raises ------ GetParameterError When the parameter provider fails to retrieve a parameter value for a given name. TransformParameterError When the parameter provider fails to transform a parameter value. Example ------- **Retrieves a parameter value from Systems Manager Parameter Store** >>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> value = get_parameter("/my/parameter") >>> >>> print(value) My parameter value **Retrieves a parameter value and decodes it using a Base64 decoder** >>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> value = get_parameter("/my/parameter", transform='binary') >>> >>> print(value) My parameter value """ # Only create the provider if this function is called at least once if "ssm" not in DEFAULT_PROVIDERS: DEFAULT_PROVIDERS["ssm"] = SSMProvider() # Add to `decrypt` sdk_options to we can have an explicit option for this sdk_options["decrypt"] = decrypt return DEFAULT_PROVIDERS["ssm"].get(name, transform=transform, force_fetch=force_fetch, **sdk_options)
def get_parameters(path: str, transform: Union[str, NoneType] = None, recursive: bool = True, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]
-
Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store
Parameters
path
:str
- Path to retrieve the parameters
transform
:str
, optional- Transforms the content from a JSON object ('json') or base64 binary string ('binary')
recursive
:bool
, optional- If this should retrieve the parameter values recursively or not, defaults to True
decrypt
:bool
, optional- If the parameter values should be decrypted
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
sdk_options
:dict
, optional- Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call
Raises
GetParameterError
- When the parameter provider fails to retrieve parameter values for a given path.
TransformParameterError
- When the parameter provider fails to transform a parameter value.
Example
Retrieves parameter values from Systems Manager Parameter Store
>>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> values = get_parameters("/my/path/prefix") >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c
Retrieves parameter values and decodes them using a Base64 decoder
>>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> values = get_parameters("/my/path/prefix", transform='binary')
Expand source code
def get_parameters( path: str, transform: Optional[str] = None, recursive: bool = True, decrypt: bool = False, force_fetch: bool = False, **sdk_options ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]: """ Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store Parameters ---------- path: str Path to retrieve the parameters transform: str, optional Transforms the content from a JSON object ('json') or base64 binary string ('binary') recursive: bool, optional If this should retrieve the parameter values recursively or not, defaults to True decrypt: bool, optional If the parameter values should be decrypted force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call Raises ------ GetParameterError When the parameter provider fails to retrieve parameter values for a given path. TransformParameterError When the parameter provider fails to transform a parameter value. Example ------- **Retrieves parameter values from Systems Manager Parameter Store** >>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> values = get_parameters("/my/path/prefix") >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c **Retrieves parameter values and decodes them using a Base64 decoder** >>> from aws_lambda_powertools.utilities.parameters import get_parameter >>> >>> values = get_parameters("/my/path/prefix", transform='binary') """ # Only create the provider if this function is called at least once if "ssm" not in DEFAULT_PROVIDERS: DEFAULT_PROVIDERS["ssm"] = SSMProvider() sdk_options["recursive"] = recursive sdk_options["decrypt"] = decrypt return DEFAULT_PROVIDERS["ssm"].get_multiple(path, transform=transform, force_fetch=force_fetch, **sdk_options)
Classes
class SSMProvider (config: Union[botocore.config.Config, NoneType] = None)
-
AWS Systems Manager Parameter Store Provider
Parameters
config
:botocore.config.Config
, optional- Botocore configuration to pass during client initialization
Example
Retrieves a parameter value from Systems Manager Parameter Store
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> value = ssm_provider.get("/my/parameter") >>> >>> print(value) My parameter value
Retrieves a parameter value from Systems Manager Parameter Store in another AWS region
>>> from botocore.config import Config >>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> >>> config = Config(region_name="us-west-1") >>> ssm_provider = SSMProvider(config=config) >>> >>> value = ssm_provider.get("/my/parameter") >>> >>> print(value) My parameter value
Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> values = ssm_provider.get_multiple("/my/path/prefix") >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c
Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call
>>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10) >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c
Initialize the SSM Parameter Store client
Expand source code
class SSMProvider(BaseProvider): """ AWS Systems Manager Parameter Store Provider Parameters ---------- config: botocore.config.Config, optional Botocore configuration to pass during client initialization Example ------- **Retrieves a parameter value from Systems Manager Parameter Store** >>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> value = ssm_provider.get("/my/parameter") >>> >>> print(value) My parameter value **Retrieves a parameter value from Systems Manager Parameter Store in another AWS region** >>> from botocore.config import Config >>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> >>> config = Config(region_name="us-west-1") >>> ssm_provider = SSMProvider(config=config) >>> >>> value = ssm_provider.get("/my/parameter") >>> >>> print(value) My parameter value **Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix** >>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> values = ssm_provider.get_multiple("/my/path/prefix") >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c **Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call** >>> from aws_lambda_powertools.utilities.parameters import SSMProvider >>> ssm_provider = SSMProvider() >>> >>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10) >>> >>> for key, value in values.items(): ... print(key, value) /my/path/prefix/a Parameter value a /my/path/prefix/b Parameter value b /my/path/prefix/c Parameter value c """ client: Any = None def __init__(self, config: Optional[Config] = None): """ Initialize the SSM Parameter Store client """ config = config or Config() self.client = boto3.client("ssm", config=config) super().__init__() def get( self, name: str, max_age: int = DEFAULT_MAX_AGE_SECS, transform: Optional[str] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options ) -> Union[str, list, dict, bytes]: """ Retrieve a parameter value or return the cached value Parameters ---------- name: str Parameter name max_age: int Maximum age of the cached value transform: str Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values. decrypt: bool, optional If the parameter value should be decrypted force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Arguments that will be passed directly to the underlying API call Raises ------ GetParameterError When the parameter provider fails to retrieve a parameter value for a given name. TransformParameterError When the parameter provider fails to transform a parameter value. """ # Add to `decrypt` sdk_options to we can have an explicit option for this sdk_options["decrypt"] = decrypt return super().get(name, max_age, transform, force_fetch, **sdk_options) def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str: """ Retrieve a parameter value from AWS Systems Manager Parameter Store Parameters ---------- name: str Parameter name decrypt: bool, optional If the parameter value should be decrypted sdk_options: dict, optional Dictionary of options that will be passed to the Parameter Store get_parameter API call """ # Explicit arguments will take precedence over keyword arguments sdk_options["Name"] = name sdk_options["WithDecryption"] = decrypt return self.client.get_parameter(**sdk_options)["Parameter"]["Value"] def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = False, **sdk_options) -> Dict[str, str]: """ Retrieve multiple parameter values from AWS Systems Manager Parameter Store Parameters ---------- path: str Path to retrieve the parameters decrypt: bool, optional If the parameter values should be decrypted recursive: bool, optional If this should retrieve the parameter values recursively or not sdk_options: dict, optional Dictionary of options that will be passed to the Parameter Store get_parameters_by_path API call """ # Explicit arguments will take precedence over keyword arguments sdk_options["Path"] = path sdk_options["WithDecryption"] = decrypt sdk_options["Recursive"] = recursive parameters = {} for page in self.client.get_paginator("get_parameters_by_path").paginate(**sdk_options): for parameter in page.get("Parameters", []): # Standardize the parameter name # The parameter name returned by SSM will contained the full path. # However, for readability, we should return only the part after # the path. name = parameter["Name"] if name.startswith(path): name = name[len(path) :] name = name.lstrip("/") parameters[name] = parameter["Value"] return parameters
Ancestors
- BaseProvider
- abc.ABC
Class variables
var client : Any
Methods
def get(self, name: str, max_age: int = 5, transform: Union[str, NoneType] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options) ‑> Union[str, list, dict, bytes]
-
Retrieve a parameter value or return the cached value
Parameters
name
:str
- Parameter name
max_age
:int
- Maximum age of the cached value
transform
:str
- Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values.
decrypt
:bool
, optional- If the parameter value should be decrypted
force_fetch
:bool
, optional- Force update even before a cached item has expired, defaults to False
sdk_options
:dict
, optional- Arguments that will be passed directly to the underlying API call
Raises
GetParameterError
- When the parameter provider fails to retrieve a parameter value for a given name.
TransformParameterError
- When the parameter provider fails to transform a parameter value.
Expand source code
def get( self, name: str, max_age: int = DEFAULT_MAX_AGE_SECS, transform: Optional[str] = None, decrypt: bool = False, force_fetch: bool = False, **sdk_options ) -> Union[str, list, dict, bytes]: """ Retrieve a parameter value or return the cached value Parameters ---------- name: str Parameter name max_age: int Maximum age of the cached value transform: str Optional transformation of the parameter value. Supported values are "json" for JSON strings and "binary" for base 64 encoded values. decrypt: bool, optional If the parameter value should be decrypted force_fetch: bool, optional Force update even before a cached item has expired, defaults to False sdk_options: dict, optional Arguments that will be passed directly to the underlying API call Raises ------ GetParameterError When the parameter provider fails to retrieve a parameter value for a given name. TransformParameterError When the parameter provider fails to transform a parameter value. """ # Add to `decrypt` sdk_options to we can have an explicit option for this sdk_options["decrypt"] = decrypt return super().get(name, max_age, transform, force_fetch, **sdk_options)
Inherited members