)
This module wraps the AWS Parameter Store and adds a caching and grouping layer with max-age invalidation.
You can use this module with AWS Lambda to read and refresh parameters and secrets. Your IAM role will require ssm:GetParameters permissions (optionally, also kms:Decrypt if you use SecureString params).
Install the module with pip:
pip install ssm-cache
A single parameter, configured by name.
from ssm_cache import SSMParameter
param = SSMParameter('my_param_name')
value = param.value
You can configure the max_age in seconds, after which the values will be automatically refreshed.
from ssm_cache import SSMParameter
param_1 = SSMParameter('param_1', max_age=300) # 5 min
value_1 = param.value
param_2 = SSMParameter('param_2', max_age=3600) # 1 hour
value_2 = param_2.value
You can configure more than one parameter to be fetched/cached/decrypted as a group.
from ssm_cache import SSMParameterGroup
group = SSMParameterGroup(max_age=300)
param_1 = group.parameter('param_1')
param_2 = group.parameter('param_2')
value_1 = param_1.value
value_2 = param_2.value
You can fetch/cache a group of parameters under a given prefix. Optionally, the group itself could have its own base path.
from ssm_cache import SSMParameterGroup
group = SSMParameterGroup(base_path="/Foo")
foo_bar = group.parameter('/Bar') # will fetch /Foo/Bar
baz_params = group.parameters('/Baz') # will fetch /Foo/Baz/1 and /Foo/Baz/2
assert len(group) == 3
Note: you can call group.parameters(...) multiple times. If caching is enabled, the group's cache will expire when the firstly fetched parameters expire.
You can filter by parameter Type and KMS KeyId, either building the filter object manually or using a class-based approach (which provides some additional checks before invoking the API).
from ssm_cache import SSMParameterGroup
from ssm_cache.filters import SSMFilterType
group = SSMParameterGroup()
# manual filter definition
params = group.parameters(
path="/Foo/Bar",
filters=[{
'Key': 'Type',
'Option': 'Equals',
'Values': ['StringList']
}],
)
# class-based filter
params = group.parameters(
path="/Foo/Bar",
filters=[SSMFilterType().value('StringList')], # will validate allowed value(s)
)
You can disable recursion when fetching parameters via prefix.
from ssm_cache import SSMParameterGroup
group = SSMParameterGroup()
# will fetch /Foo/1, but not /Foo/Bar/1
params = group.parameters(
path="/Foo",
recursive=False,
)
StringList parameters (documentation here) are automatically converted to Python lists with no additional configuration.
from ssm_cache import SSMParameter
# "my_twitter_api_keys" is a StringList parameter (four comma-separated values)
twitter_params = SSMParameter('my_twitter_api_keys')
key, secret, access_token, access_token_secret = twitter_params.value
You can manually force a refresh on a parameter or parameter group. Note that if a parameter is part of a group, the refresh operation will involve the entire group.
from ssm_cache import SSMParameter
param = SSMParameter('my_param_name')
value = param.value
param.refresh()
new_value = param.value
from ssm_cache import SSMParameterGroup
group = SSMParameterGroup()
param_1 = group.parameter('param_1')
param_2 = group.parameter('param_2')
value_1 = param_1.value
value_2 = param_2.value
group.refresh()
new_value_1 = param_1.value
new_value_2 = param_2.value
param_1.refresh()
new_new_value_1 = param_1.value
new_new_value_2 = param_2.value # one parameter refreshes the whole group
Decryption is enabled by default, but you can explicitly disable it (works for SSMParameter and SSMGroup).
from ssm_cache import SSMParameter
param = SSMParameter('my_param_name', with_decryption=False)
value = param.value
You can read AWS Secrets Manager secrets transparently by using the SecretsManagerParameter class, which comes with the same interface of SSMParameter and performs some additional prefixing and validation.
from ssm_cache import SecretsManagerParameter
secret = SecretsManagerParameter('my_secret_name')
value = secret.value
Secrets can be added to a SSMParameterGroup as well, although no group prefix will be applied.
from ssm_cache import SSMParameterGroup
group = SSMParameterGroup()
param = group.parameter('my_param')
secret = group.secret('my_secret')
param_value = param.value
secret_value = secret.value
SSM Parameter Store supports version selectors (documentation here).
By default, the latest version is fetched if you don't specify it.
Here is how you can retrieve a specific parameter version:
from ssm_cache import SSMParameter
param = SSMParameter('my_param_name:2')
value = param.value
Please note that invoking param.refresh() will not fetch newer versions. This is the intended behavior, as version selection should be used only when you need a specific parameter version.
If you don't specify any version, you can always read the current version of a parameter. In this case, invoking param.refresh() will invoke the new version.
from ssm_cache import SSMParameter
param = SSMParameter('my_param_name')
print(param.version) # will print an int
Your AWS Lambda code will look similar to the following snippet.
from ssm_cache import SSMParameter, SecretsManagerParameter
param = SSMParameter('my_param_name')
secret = SecretsManagerParameter('my_secret_name')
def lambda_handler(event, context):
dbname = param.value
password = secret.value
return 'Hello from Lambda with dbname %s and password %s' % (dbname, password)
You may want to explicitly refresh the parameter cache when you believe the cached value expired.
In the example below, we refresh the parameter value when an InvalidCredentials exception is detected (see the decorator utility for a simpler version!).
from ssm_cache import SSMParameter
from my_db_lib import Client, InvalidCredentials # pseudo-code
param = SSMParameter('my_db_password')
my_db_client = Client(password=param.value)
def read_record(is_retry=False):
try:
return my_db_client.read_record()
except InvalidCredentials:
if not is_retry: # avoid infinite recursion
param.refresh() # force parameter refresh
my_db_client = Client(password=param.value) # re-configure db client
return read_record(is_retry=True) # let's try again :)
def lambda_handler(event, context):
return {
'record': read_record(),
}
The retry logic shown above can be simplified with the decorator method provided by each SSMParameter and SSMParameterGroup object.
The @refresh_on_error decorator will intercept errors (or a specific error_class, if given), refresh the parameters values, and attempt to re-call the decorated function. Optionally, you can provide a callback argument to implement your own logic (in the example below, to create a new db client with the new password).
from ssm_cache import SSMParameter
from my_db_lib import Client, InvalidCredentials # pseudo-code
param = SSMParameter('my_db_password')
my_db_client = Client(password=param.value)
def on_error_callback():
my_db_client = Client(password=param.value)
@param.refresh_on_error(InvalidCredentials, on_error_callback)
def read_record(is_retry=False):
return my_db_client.read_record()
def lambda_handler(event, context):
return {
'record': read_record(),
}
The refresh_on_error decorator supports the following arguments:
Exception)None)"is_retry")If you want to replace the default boto3 SSM client, SSMParameter allows you to call set_ssm_client and provide your own boto3 client or even a custom object. Note that such custom object will need to implement two methods: get_parameters and get_parameters_by_path.
For example, here's how you could inject a Placebo client for local tests:
import placebo, boto3
from ssm_cache import SSMParameter
# create regular boto3 session
session = boto3.Session()
# attach placebo to the session
pill = placebo.attach(session, data_path=PLACEBO_PATH)
pill.playback()
# create special boto3 client
client = session.client('ssm')
# inject special client into SSMParameter or SSMParameterGroup
SSMParameter.set_ssm_client(client)
Clone this repository, create a virtualenv and install all the dev dependencies:
git clone https://github.com/alexcasalboni/ssm-cache-python.git
cd ssm-cache-python
virtualenv env
source env/bin/activate
pip install -r requirements-dev.txt
You can run tests with nose:
nosetests
Generate a coverage report:
nosetests --with-coverage --cover-erase --cover-html --cover-package=ssm_cache
open cover/index.html
Run pylint:
pylint ssm_cache
Note: when you open a new PR, GitHub will run tests on multiple Python environments and verify the new coverage for you, but we highly recommend you run these tasks locally as well before submitting new code.
SSMParameterGroup supportFAQs
AWS System Manager Parameter Store caching client for Python
We found that ssm-cache demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.