Module sls_tools.param_store.param_store
Expand source code
import os
import boto3
import logging
from enum import Enum
from .param_store_result import ParamStoreResult
class ParamStore:
"""Provides access to key/values (variables) stored locally on the OS or AWS SSM.
examples:
# Set and get a value in SSM:
ParamStore.set('MY_VAR', '123')
my_var = ParamStore.get('MY_VAR').value
>> '123'
# Set and get a value from the OS:
ParamStore.set('MY_VAR', '123', store=ParamStore.Stores.OS)
my_var = ParamStore.get('MY_VAR').value
>> '123'
# Or explicitly:
my_var = ParamStore.get('MY_VAR', store=ParamStore.Stores.OS).value
>> '123'
# Set and get a value in SSM and convert it to a list:
ParamStore.set('MY_VAR', 'one,two,three')
my_var = ParamStore.get('MY_VAR').to_list()
>> ['one', 'two', 'three']
"""
class Stores(Enum):
"""Enum for the available Key/Value stores.
Supported options are: OS, SSM, AUTO
"AUTO" is a reserved keyword that can be passed to each method to delegate
the choice to each individual method.
"""
OS = 'os'
SSM = 'ssm'
AUTO = 'auto'
@classmethod
def set(cls, key, value, store=Stores.SSM):
"""Sets a key/value pair.
Args:
key: The key to set.
value: The value to set.
store: Where to set the value. Defaults to Stores.SSM.
Returns:
True if successfully set otherwise False.
"""
result = False
# Set the value in the local environment.
if store in [cls.Stores.OS, cls.Stores.AUTO]:
result = cls._set_in_os(key, value)
# Set the value in SSM.
if store in [cls.Stores.SSM, cls.Stores.AUTO]:
result = cls._set_in_ssm(key, value)
return result
@classmethod
def get(cls, key, default=None, store=Stores.AUTO):
"""Gets the value for a key.
Will return the value from the OS if its set otherwise it will try to get the value from SSM.
Args:
key: The key for the value to get.
default: The value to return if the key value is None.
store: Where to get the value from. Defaults to Stores.AUTO.
Returns:
ParamStoreResult
"""
# Try to get the value from the local environment.
if store in [cls.Stores.OS, cls.Stores.AUTO]:
result = cls._get_from_os(key)
if result is not None:
return ParamStoreResult(key, result, cls.Stores.OS)
# Try to get the value from SSM.
if store in [cls.Stores.SSM, cls.Stores.AUTO]:
result = cls._get_from_ssm(key)
if result is not None:
return ParamStoreResult(key, result, cls.Stores.SSM)
return ParamStoreResult(key, default, None)
@classmethod
def delete(cls, key, store=Stores.AUTO):
"""Deletes a key/value.
Args:
key: The key to delete.
store: Where to delete from. Defaults to Stores.AUTO.
Returns:
True if successfully deleted otherwise False.
"""
result = False
# Delete the value from the local environment.
if store in [cls.Stores.OS, cls.Stores.AUTO]:
result = cls._delete_from_os(key)
# Delete the value from SSM.
if store in [cls.Stores.SSM, cls.Stores.AUTO]:
result = cls._delete_from_ssm(key)
return result
@classmethod
def contains(cls, key, store=Stores.AUTO):
"""Gets if a key is present in one of the param stores.
Args:
key: The key to check for.
store: Where to look for the key. Defaults to Stores.AUTO.
Returns:
True if the key is contained in the store.
"""
result = cls.get(key, default=None, store=store)
return result.store is not None and result.value is not None
@classmethod
def _get_from_os(cls, key):
"""Gets a value from the OS environment variables.
Args:
key: The key for the value to get.
Returns:
The key value or None.
"""
return os.environ.get(key)
@classmethod
def _set_in_os(cls, key, value):
"""Sets a key/value in the OS environment.
Args:
key: The key to set.
value: The value to set.
Returns:
True
"""
os.environ[key] = value
return True
@classmethod
def _delete_from_os(cls, key):
"""Deletes the key from the OS environment.
Args:
key: The key to delete.
Returns:
True
"""
if key in os.environ.keys():
del os.environ[key]
return True
@classmethod
def _get_from_ssm(cls, key):
"""Gets a value from SSM.
Args:
key: The key for the value to get.
Returns:
The key value or None, or raises
"""
result = None
ssm_key = cls._build_ssm_key(key)
client = cls._get_ssm_client()
try:
get_response = client.get_parameter(Name=ssm_key, WithDecryption=True)
result = get_response.get('Parameter').get('Value')
except client.exceptions.ParameterNotFound:
logging.exception('SSM Parameter Not Found: {0}'.format(ssm_key))
except Exception as ex:
logging.exception('SSM Error: {0}'.format(ex))
return result
@classmethod
def _set_in_ssm(cls, key, value, type='SecureString'):
"""Sets a key/value pair in SSM.
Args:
key: The key to set.
value: The value to set.
type: The SSM parameter type. Defaults to SecureString.
Returns:
True if successful otherwise False.
"""
ssm_key = cls._build_ssm_key(key)
try:
cls._get_ssm_client().put_parameter(Name=ssm_key, Value=value, Type=type, Overwrite=True)
return True
except Exception as ex:
logging.exception('SSM Error: {0}'.format(ex))
return False
@classmethod
def _delete_from_ssm(cls, key):
"""Deletes a key in SSM.
Args:
key: The key to delete.
Returns:
True if successful or the key does not exist otherwise False.
"""
ssm_key = cls._build_ssm_key(key)
try:
cls._get_ssm_client().delete_parameter(Name=ssm_key)
return True
except cls._get_ssm_client().exceptions.ParameterNotFound as ex:
return True
except Exception as ex:
logging.exception('SSM Error: {0}'.format(ex))
return False
@classmethod
def _build_ssm_key(cls, key):
"""Builds an SSM key.
The format is "service_name/service_stage/key".
Args:
key: The key to build a fully qualified key for.
Returns:
The fully qualified key string.
"""
service_name = cls.get('SERVICE_NAME', store=cls.Stores.OS).value
service_stage = cls.get('SERVICE_STAGE', store=cls.Stores.OS).value
if not service_name:
raise ValueError('Environment variable not set: SERVICE_NAME')
if not service_stage:
raise ValueError('Environment variable not set: SERVICE_STAGE')
return '/{0}/{1}/{2}'.format(service_name, service_stage, key)
# Cached instance of the SSM client.
_ssm_client = None
@classmethod
def _get_ssm_client(cls):
"""Gets an SSM boto3 client.
Returns:
SSM client or None.
"""
try:
if not cls._ssm_client:
cls._ssm_client = boto3.client('ssm')
return cls._ssm_client
except Exception as ex:
logging.exception('SSM Error: {0}'.format(ex))
return None
Classes
class ParamStore
-
Provides access to key/values (variables) stored locally on the OS or AWS SSM.
examples:
# Set and get a value in SSM: ParamStore.set('MY_VAR', '123') my_var = ParamStore.get('MY_VAR').value >> '123' # Set and get a value from the OS: ParamStore.set('MY_VAR', '123', store=ParamStore.Stores.OS) my_var = ParamStore.get('MY_VAR').value >> '123' # Or explicitly: my_var = ParamStore.get('MY_VAR', store=ParamStore.Stores.OS).value >> '123' # Set and get a value in SSM and convert it to a list: ParamStore.set('MY_VAR', 'one,two,three') my_var = ParamStore.get('MY_VAR').to_list() >> ['one', 'two', 'three']
Expand source code
class ParamStore: """Provides access to key/values (variables) stored locally on the OS or AWS SSM. examples: # Set and get a value in SSM: ParamStore.set('MY_VAR', '123') my_var = ParamStore.get('MY_VAR').value >> '123' # Set and get a value from the OS: ParamStore.set('MY_VAR', '123', store=ParamStore.Stores.OS) my_var = ParamStore.get('MY_VAR').value >> '123' # Or explicitly: my_var = ParamStore.get('MY_VAR', store=ParamStore.Stores.OS).value >> '123' # Set and get a value in SSM and convert it to a list: ParamStore.set('MY_VAR', 'one,two,three') my_var = ParamStore.get('MY_VAR').to_list() >> ['one', 'two', 'three'] """ class Stores(Enum): """Enum for the available Key/Value stores. Supported options are: OS, SSM, AUTO "AUTO" is a reserved keyword that can be passed to each method to delegate the choice to each individual method. """ OS = 'os' SSM = 'ssm' AUTO = 'auto' @classmethod def set(cls, key, value, store=Stores.SSM): """Sets a key/value pair. Args: key: The key to set. value: The value to set. store: Where to set the value. Defaults to Stores.SSM. Returns: True if successfully set otherwise False. """ result = False # Set the value in the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._set_in_os(key, value) # Set the value in SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._set_in_ssm(key, value) return result @classmethod def get(cls, key, default=None, store=Stores.AUTO): """Gets the value for a key. Will return the value from the OS if its set otherwise it will try to get the value from SSM. Args: key: The key for the value to get. default: The value to return if the key value is None. store: Where to get the value from. Defaults to Stores.AUTO. Returns: ParamStoreResult """ # Try to get the value from the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._get_from_os(key) if result is not None: return ParamStoreResult(key, result, cls.Stores.OS) # Try to get the value from SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._get_from_ssm(key) if result is not None: return ParamStoreResult(key, result, cls.Stores.SSM) return ParamStoreResult(key, default, None) @classmethod def delete(cls, key, store=Stores.AUTO): """Deletes a key/value. Args: key: The key to delete. store: Where to delete from. Defaults to Stores.AUTO. Returns: True if successfully deleted otherwise False. """ result = False # Delete the value from the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._delete_from_os(key) # Delete the value from SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._delete_from_ssm(key) return result @classmethod def contains(cls, key, store=Stores.AUTO): """Gets if a key is present in one of the param stores. Args: key: The key to check for. store: Where to look for the key. Defaults to Stores.AUTO. Returns: True if the key is contained in the store. """ result = cls.get(key, default=None, store=store) return result.store is not None and result.value is not None @classmethod def _get_from_os(cls, key): """Gets a value from the OS environment variables. Args: key: The key for the value to get. Returns: The key value or None. """ return os.environ.get(key) @classmethod def _set_in_os(cls, key, value): """Sets a key/value in the OS environment. Args: key: The key to set. value: The value to set. Returns: True """ os.environ[key] = value return True @classmethod def _delete_from_os(cls, key): """Deletes the key from the OS environment. Args: key: The key to delete. Returns: True """ if key in os.environ.keys(): del os.environ[key] return True @classmethod def _get_from_ssm(cls, key): """Gets a value from SSM. Args: key: The key for the value to get. Returns: The key value or None, or raises """ result = None ssm_key = cls._build_ssm_key(key) client = cls._get_ssm_client() try: get_response = client.get_parameter(Name=ssm_key, WithDecryption=True) result = get_response.get('Parameter').get('Value') except client.exceptions.ParameterNotFound: logging.exception('SSM Parameter Not Found: {0}'.format(ssm_key)) except Exception as ex: logging.exception('SSM Error: {0}'.format(ex)) return result @classmethod def _set_in_ssm(cls, key, value, type='SecureString'): """Sets a key/value pair in SSM. Args: key: The key to set. value: The value to set. type: The SSM parameter type. Defaults to SecureString. Returns: True if successful otherwise False. """ ssm_key = cls._build_ssm_key(key) try: cls._get_ssm_client().put_parameter(Name=ssm_key, Value=value, Type=type, Overwrite=True) return True except Exception as ex: logging.exception('SSM Error: {0}'.format(ex)) return False @classmethod def _delete_from_ssm(cls, key): """Deletes a key in SSM. Args: key: The key to delete. Returns: True if successful or the key does not exist otherwise False. """ ssm_key = cls._build_ssm_key(key) try: cls._get_ssm_client().delete_parameter(Name=ssm_key) return True except cls._get_ssm_client().exceptions.ParameterNotFound as ex: return True except Exception as ex: logging.exception('SSM Error: {0}'.format(ex)) return False @classmethod def _build_ssm_key(cls, key): """Builds an SSM key. The format is "service_name/service_stage/key". Args: key: The key to build a fully qualified key for. Returns: The fully qualified key string. """ service_name = cls.get('SERVICE_NAME', store=cls.Stores.OS).value service_stage = cls.get('SERVICE_STAGE', store=cls.Stores.OS).value if not service_name: raise ValueError('Environment variable not set: SERVICE_NAME') if not service_stage: raise ValueError('Environment variable not set: SERVICE_STAGE') return '/{0}/{1}/{2}'.format(service_name, service_stage, key) # Cached instance of the SSM client. _ssm_client = None @classmethod def _get_ssm_client(cls): """Gets an SSM boto3 client. Returns: SSM client or None. """ try: if not cls._ssm_client: cls._ssm_client = boto3.client('ssm') return cls._ssm_client except Exception as ex: logging.exception('SSM Error: {0}'.format(ex)) return None
Class variables
var Stores
-
Enum for the available Key/Value stores.
Supported options are: OS, SSM, AUTO
"AUTO" is a reserved keyword that can be passed to each method to delegate the choice to each individual method.
Static methods
def contains(key, store=Stores.AUTO)
-
Gets if a key is present in one of the param stores.
Args
key
- The key to check for.
store
- Where to look for the key. Defaults to Stores.AUTO.
Returns
True if the key is contained in the store.
Expand source code
@classmethod def contains(cls, key, store=Stores.AUTO): """Gets if a key is present in one of the param stores. Args: key: The key to check for. store: Where to look for the key. Defaults to Stores.AUTO. Returns: True if the key is contained in the store. """ result = cls.get(key, default=None, store=store) return result.store is not None and result.value is not None
def delete(key, store=Stores.AUTO)
-
Deletes a key/value.
Args
key
- The key to delete.
store
- Where to delete from. Defaults to Stores.AUTO.
Returns
True if successfully deleted otherwise False.
Expand source code
@classmethod def delete(cls, key, store=Stores.AUTO): """Deletes a key/value. Args: key: The key to delete. store: Where to delete from. Defaults to Stores.AUTO. Returns: True if successfully deleted otherwise False. """ result = False # Delete the value from the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._delete_from_os(key) # Delete the value from SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._delete_from_ssm(key) return result
def get(key, default=None, store=Stores.AUTO)
-
Gets the value for a key.
Will return the value from the OS if its set otherwise it will try to get the value from SSM.
Args
key
- The key for the value to get.
default
- The value to return if the key value is None.
store
- Where to get the value from. Defaults to Stores.AUTO.
Returns
ParamStoreResult
Expand source code
@classmethod def get(cls, key, default=None, store=Stores.AUTO): """Gets the value for a key. Will return the value from the OS if its set otherwise it will try to get the value from SSM. Args: key: The key for the value to get. default: The value to return if the key value is None. store: Where to get the value from. Defaults to Stores.AUTO. Returns: ParamStoreResult """ # Try to get the value from the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._get_from_os(key) if result is not None: return ParamStoreResult(key, result, cls.Stores.OS) # Try to get the value from SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._get_from_ssm(key) if result is not None: return ParamStoreResult(key, result, cls.Stores.SSM) return ParamStoreResult(key, default, None)
def set(key, value, store=Stores.SSM)
-
Sets a key/value pair.
Args
key
- The key to set.
value
- The value to set.
store
- Where to set the value. Defaults to Stores.SSM.
Returns
True if successfully set otherwise False.
Expand source code
@classmethod def set(cls, key, value, store=Stores.SSM): """Sets a key/value pair. Args: key: The key to set. value: The value to set. store: Where to set the value. Defaults to Stores.SSM. Returns: True if successfully set otherwise False. """ result = False # Set the value in the local environment. if store in [cls.Stores.OS, cls.Stores.AUTO]: result = cls._set_in_os(key, value) # Set the value in SSM. if store in [cls.Stores.SSM, cls.Stores.AUTO]: result = cls._set_in_ssm(key, value) return result