Source code for betterboto.ssm
import types
import logging
import time
logger = logging.getLogger(__file__)
put_parameter_and_wait_max_retries = 10
[docs]def put_parameter_and_wait(self, Name, **kwargs):
"""
This will call put_parameter and ensure it has been successfully put by checking it
:param self: ssm client
:param Name: The fully qualified name of the parameter that you want to add to the system
:param kwargs: These args are passed through to ssm.put_parameter
:return: ssm.get_parameter.response
"""
new_version = self.put_parameter(Name=Name, **kwargs).get('Version')
count = 0
current_parameter = {
"Parameter": {
"Version": -1
}
}
while current_parameter.get('Parameter').get('Version') < new_version:
count += 1
if count > put_parameter_and_wait_max_retries:
raise Exception(f"Putting and waiting for param {Name} failed")
time.sleep(1)
if new_version == 1:
try:
current_parameter = self.get_parameter(Name=Name)
except self.exceptions.ParameterNotFound:
pass
else:
current_parameter = self.get_parameter(Name=Name)
return current_parameter
[docs]def get_parameter_history_single_page(self, **kwargs):
"""
This will continue to call get_parameter_history_single_page until there are no more pages left to retrieve.
It will return the aggregated response in the same structure as get_parameter_history_single_page does.
:param self: ssm client
:param kwargs: these are passed onto the get_parameter_history_single_page method call
:return: ssm_client.get_parameter_history_single_page.response
"""
return slurp(
'get_parameter_history',
self.get_parameter_history,
'Parameters',
'NextToken', 'NextToken',
**kwargs
)
[docs]class ParameterVersionNotFoundException(Exception):
pass
def get_parameter_version(self, Version, **kwargs):
paginator = self.get_paginator('get_parameter_history')
iterator = paginator.paginate(
**kwargs
)
for page in iterator:
for parameter in page.get("Parameters", []):
if parameter.get("Version") == Version:
return dict(Parameter=parameter)
raise ParameterVersionNotFoundException(f"Could not find version: {Version} of {kwargs.get('Name')}")
def make_better(client):
client.put_parameter_and_wait = types.MethodType(put_parameter_and_wait, client)
client.get_parameter_history_single_page = types.MethodType(get_parameter_history_single_page, client)
client.get_parameter_version = types.MethodType(get_parameter_version, client)
return client