-
-
Save cloudfiles-me/b4c44e28c8cdbdb5cc476c0fd9f26a98 to your computer and use it in GitHub Desktop.
Python class to provide a dictionary-like interface to access AWS SSM Parameter Store easily
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import boto3 | |
| from botocore.exceptions import ClientError | |
| import datetime | |
| class SSMParameterStore(object): | |
| """ | |
| Provide a dictionary-like interface to access AWS SSM Parameter Store | |
| """ | |
| def __init__(self, prefix=None, ssm_client=None, ttl=None): | |
| self._prefix = (prefix or "").rstrip("/") + "/" | |
| self._client = ssm_client or boto3.client('ssm') | |
| self._keys = None | |
| self._substores = {} | |
| self._ttl = ttl | |
| def get(self, name, **kwargs): | |
| assert name, 'Name can not be empty' | |
| if self._keys is None: | |
| self.refresh() | |
| abs_key = "%s%s" % (self._prefix, name) | |
| if name not in self._keys: | |
| if 'default' in kwargs: | |
| return kwargs['default'] | |
| raise KeyError(name) | |
| elif self._keys[name]['type'] == 'prefix': | |
| if abs_key not in self._substores: | |
| store = self.__class__(prefix=abs_key, ssm_client=self._client, ttl=self._ttl) | |
| self._substores[abs_key] = store | |
| return self._substores[abs_key] | |
| else: | |
| return self._get_value(name, abs_key) | |
| def refresh(self): | |
| self._keys = {} | |
| self._substores = {} | |
| responses = self._client.describe_parameters( | |
| ParameterFilters=[ | |
| dict(Key="Path", Option="Recursive", Values=[self._prefix]) | |
| ] | |
| ) | |
| for p in responses[u'Parameters']: | |
| paths = p['Name'][len(self._prefix):].split('/') | |
| name = paths[0] | |
| # this is a prefix | |
| if len(paths) > 1: | |
| self._keys[name] = {'type': 'prefix'} | |
| else: | |
| self._keys[name] = {'type': 'parameter', 'expire': None} | |
| def keys(self): | |
| if self._keys is None: | |
| self.refresh() | |
| return self._keys.keys() | |
| def _get_value(self, name, abs_key): | |
| entry = self._keys[name] | |
| # simple ttl | |
| if self._ttl == False or (entry['expire'] and entry['expire'] <= datetime.datetime.now()): | |
| entry.pop('value', None) | |
| if 'value' not in entry: | |
| parameter = self._client.get_parameter(Name=abs_key, WithDecryption=True)['Parameter'] | |
| value = parameter['Value'] | |
| if parameter['Type'] == 'StringList': | |
| value = value.split(",") | |
| entry['value'] = value | |
| if self._ttl: | |
| entry['expire'] = datetime.datetime.now() + datetime.timedelta(seconds=self._ttl) | |
| else: | |
| entry['expire'] = None | |
| return entry['value'] | |
| def __contains__(self, name): | |
| try: | |
| self.get(name) | |
| return True | |
| except: | |
| return False | |
| def __getitem__(self, name): | |
| return self.get(name) | |
| def __setitem__(self, key, value): | |
| raise NotImplementedError() | |
| def __delitem__(self, name): | |
| raise NotImplementedError() | |
| def __repr__(self): | |
| return 'ParameterStore[%s]' % self._prefix |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment