Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Python class to provide a dictionary-like interface to access AWS SSM Parameter Store easily
# Copyright (c) 2018 Bao Nguyen <b@nqbao.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# ==============================================================================
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)
store._keys = self._keys[name]['children']
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 = {}
paginator = self._client.get_paginator('describe_parameters')
pager = paginator.paginate(
ParameterFilters=[
dict(Key="Path", Option="Recursive", Values=[self._prefix])
]
)
for page in pager:
for p in page['Parameters']:
paths = p['Name'][len(self._prefix):].split('/')
self._update_keys(self._keys, paths)
@classmethod
def _update_keys(cls, keys, paths):
name = paths[0]
# this is a prefix
if len(paths) > 1:
if name not in keys:
keys[name] = {'type': 'prefix', 'children': {}}
cls._update_keys(keys[name]['children'], paths[1:])
else:
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
@jsharpe
Copy link

jsharpe commented Jul 26, 2018

Whats the license on this code?

@dperconti
Copy link

dperconti commented Jan 27, 2019

Nice! 👍

@geronimo-iia
Copy link

geronimo-iia commented May 21, 2019

thanks !

@dorian599
Copy link

dorian599 commented Dec 16, 2019

excellent, thanks a lot !!!

@andrewdieken
Copy link

andrewdieken commented May 15, 2020

this is great, thank you!! 🙏

@basedrhys
Copy link

basedrhys commented Jul 21, 2020

Thankyou! 💯

@mmaz2301
Copy link

mmaz2301 commented Sep 1, 2020

First of all, great job! I'm having a small issue as I'm a java guy but a Python noob. I can't seem to set a nested location. If I do this:
store = ssm_parameter_store.SSMParameterStore(prefix='/dev')
my_lambda_store = store('lambda/events)

I get an exception. If I navigate and create a store using each key, I can get what I need out of a store.

Any tips here?

tia

@nqbao
Copy link
Author

nqbao commented Sep 1, 2020

Hi @mmaz2301: You can use it as an array store['lambda']['events']

@mmaz2301
Copy link

mmaz2301 commented Sep 1, 2020

Thanks so much! I had tried that before, but kind of instinctively did ( ['lambda']['events']). Awesome piece of work!

@thoroc
Copy link

thoroc commented Feb 15, 2021

@nqbao The client exception is never used. Did you forget something?

@viotti
Copy link

viotti commented Jul 3, 2021

Using a bare except clause (line 120) is not a good practise. The better alternative is except Exception.

@DavidFelsen
Copy link

DavidFelsen commented Oct 27, 2021

@nqbao Great work, thanks for sharing!
However this won't work if there is a mixture of top-level parameters (no prefix) and subfolders, as it will cutoff the first letter of the top-level parameters. A fix on line 72 could be:

name = p['Name']
if name.startswith('/'):
    paths = name[len(self._prefix):].split('/') # Same as original version, removes the prefix
else:
    paths = name.split('/') # If there is no prefix, avoids cutting off first letter of the parameter key

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment