SageWorks Parameter Storage
Examples
Examples of using the Parameter Storage class are listed at the bottom of this page Examples.
ParameterStore: Manages SageWorks parameters in AWS Systems Manager Parameter Store.
ParameterStore
ParameterStore: Manages SageWorks parameters in AWS Systems Manager Parameter Store.
Common Usage
params = ParameterStore()
# List Parameters
params.list()
['/sageworks/abalone_info',
'/sageworks/my_data',
'/sageworks/test',
'/sageworks/pipelines/my_pipeline']
# Add Key
params.add("key", "value")
value = params.get("key")
# Add any data (lists, dictionaries, etc..)
my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
params.add("my_data", my_data)
# Retrieve data
return_value = params.get("my_data")
pprint(return_value)
{'key': 'value', 'list': [1, 2, 3], 'number': 4.2}
# Delete parameters
param_store.delete("my_data")
Source code in src/sageworks/api/parameter_store.py
| class ParameterStore:
"""ParameterStore: Manages SageWorks parameters in AWS Systems Manager Parameter Store.
Common Usage:
```python
params = ParameterStore()
# List Parameters
params.list()
['/sageworks/abalone_info',
'/sageworks/my_data',
'/sageworks/test',
'/sageworks/pipelines/my_pipeline']
# Add Key
params.add("key", "value")
value = params.get("key")
# Add any data (lists, dictionaries, etc..)
my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
params.add("my_data", my_data)
# Retrieve data
return_value = params.get("my_data")
pprint(return_value)
{'key': 'value', 'list': [1, 2, 3], 'number': 4.2}
# Delete parameters
param_store.delete("my_data")
```
"""
def __init__(self):
"""ParameterStore Init Method"""
self.log = logging.getLogger("sageworks")
# Initialize a SageWorks Session (to assume the SageWorks ExecutionRole)
self.boto3_session = AWSSession().boto3_session
# Create a Systems Manager (SSM) client for Parameter Store operations
self.ssm_client = self.boto3_session.client("ssm")
def list(self) -> list:
"""List all parameters in the AWS Parameter Store.
Returns:
list: A list of parameter names and details.
"""
try:
# Set up parameters for our search
params = {"MaxResults": 50}
# Initialize the list to collect parameter names
all_parameters = []
# Make the initial call to describe parameters
response = self.ssm_client.describe_parameters(**params)
# Aggregate the names from the initial response
all_parameters.extend(param["Name"] for param in response["Parameters"])
# Continue to paginate if there's a NextToken
while "NextToken" in response:
# Update the parameters with the NextToken for subsequent calls
params["NextToken"] = response["NextToken"]
response = self.ssm_client.describe_parameters(**params)
# Aggregate the names from the subsequent responses
all_parameters.extend(param["Name"] for param in response["Parameters"])
except Exception as e:
self.log.error(f"Failed to list parameters: {e}")
return []
# Return the aggregated list of parameter names
return all_parameters
def get(self, name: str, warn: bool = True, decrypt: bool = True) -> Union[str, list, dict, None]:
"""Retrieve a parameter value from the AWS Parameter Store.
Args:
name (str): The name of the parameter to retrieve.
warn (bool): Whether to log a warning if the parameter is not found.
decrypt (bool): Whether to decrypt secure string parameters.
Returns:
Union[str, list, dict, None]: The value of the parameter or None if not found.
"""
try:
# Retrieve the parameter from Parameter Store
response = self.ssm_client.get_parameter(Name=name, WithDecryption=decrypt)
value = response["Parameter"]["Value"]
# Auto-detect and decompress if needed
if value.startswith("COMPRESSED:"):
# Base64 decode and decompress
self.log.important(f"Decompressing parameter '{name}'...")
compressed_value = base64.b64decode(value[len("COMPRESSED:") :])
value = zlib.decompress(compressed_value).decode("utf-8")
# Attempt to parse the value back to its original type
try:
parsed_value = json.loads(value)
return parsed_value
except (json.JSONDecodeError, TypeError):
# If parsing fails, return the value as is (assumed to be a simple string)
return value
except ClientError as e:
if e.response["Error"]["Code"] == "ParameterNotFound":
if warn:
self.log.warning(f"Parameter '{name}' not found")
else:
self.log.error(f"Failed to get parameter '{name}': {e}")
return None
def add(self, name: str, value, overwrite: bool = True):
"""Add or update a parameter in the AWS Parameter Store.
Args:
name (str): The name of the parameter.
value (str | list | dict): The value of the parameter.
overwrite (bool): Whether to overwrite an existing parameter (default: True)
"""
try:
# Anything that's not a string gets converted to JSON
if not isinstance(value, str):
value = json.dumps(value)
# Check size and compress if necessary
if len(value) > 4096:
self.log.warning(f"Parameter {name} exceeds 4KB ({len(value)} Bytes) Compressing...")
compressed_value = zlib.compress(value.encode("utf-8"), level=9)
encoded_value = "COMPRESSED:" + base64.b64encode(compressed_value).decode("utf-8")
# Report on the size of the compressed value
compressed_size = len(compressed_value)
if compressed_size > 4096:
doc_link = "https://supercowpowers.github.io/sageworks/api_classes/df_store"
self.log.error(f"Compressed size {compressed_size} bytes, cannot store > 4KB")
self.log.error(f"For larger data use the DFStore() class ({doc_link})")
return
# Add or update the compressed parameter in Parameter Store
try:
# Add or update the compressed parameter in Parameter Store
self.ssm_client.put_parameter(Name=name, Value=encoded_value, Type="String", Overwrite=overwrite)
self.log.info(f"Parameter '{name}' added/updated successfully with compression.")
return
except Exception as e:
self.log.critical(f"Failed to add/update compressed parameter '{name}': {e}")
raise
# Add or update the parameter normally if under 4KB
self.ssm_client.put_parameter(Name=name, Value=value, Type="String", Overwrite=overwrite)
self.log.info(f"Parameter '{name}' added/updated successfully.")
except Exception as e:
self.log.critical(f"Failed to add/update parameter '{name}': {e}")
raise
def delete(self, name: str):
"""Delete a parameter from the AWS Parameter Store.
Args:
name (str): The name of the parameter to delete.
"""
try:
# Delete the parameter from Parameter Store
self.ssm_client.delete_parameter(Name=name)
self.log.info(f"Parameter '{name}' deleted successfully.")
except Exception as e:
self.log.error(f"Failed to delete parameter '{name}': {e}")
def __repr__(self):
"""Return a string representation of the ParameterStore object."""
return "\n".join(self.list())
|
__init__()
ParameterStore Init Method
Source code in src/sageworks/api/parameter_store.py
| def __init__(self):
"""ParameterStore Init Method"""
self.log = logging.getLogger("sageworks")
# Initialize a SageWorks Session (to assume the SageWorks ExecutionRole)
self.boto3_session = AWSSession().boto3_session
# Create a Systems Manager (SSM) client for Parameter Store operations
self.ssm_client = self.boto3_session.client("ssm")
|
__repr__()
Return a string representation of the ParameterStore object.
Source code in src/sageworks/api/parameter_store.py
| def __repr__(self):
"""Return a string representation of the ParameterStore object."""
return "\n".join(self.list())
|
add(name, value, overwrite=True)
Add or update a parameter in the AWS Parameter Store.
Parameters:
Name |
Type |
Description |
Default |
name |
str
|
The name of the parameter.
|
required
|
value |
str | list | dict
|
The value of the parameter.
|
required
|
overwrite |
bool
|
Whether to overwrite an existing parameter (default: True)
|
True
|
Source code in src/sageworks/api/parameter_store.py
| def add(self, name: str, value, overwrite: bool = True):
"""Add or update a parameter in the AWS Parameter Store.
Args:
name (str): The name of the parameter.
value (str | list | dict): The value of the parameter.
overwrite (bool): Whether to overwrite an existing parameter (default: True)
"""
try:
# Anything that's not a string gets converted to JSON
if not isinstance(value, str):
value = json.dumps(value)
# Check size and compress if necessary
if len(value) > 4096:
self.log.warning(f"Parameter {name} exceeds 4KB ({len(value)} Bytes) Compressing...")
compressed_value = zlib.compress(value.encode("utf-8"), level=9)
encoded_value = "COMPRESSED:" + base64.b64encode(compressed_value).decode("utf-8")
# Report on the size of the compressed value
compressed_size = len(compressed_value)
if compressed_size > 4096:
doc_link = "https://supercowpowers.github.io/sageworks/api_classes/df_store"
self.log.error(f"Compressed size {compressed_size} bytes, cannot store > 4KB")
self.log.error(f"For larger data use the DFStore() class ({doc_link})")
return
# Add or update the compressed parameter in Parameter Store
try:
# Add or update the compressed parameter in Parameter Store
self.ssm_client.put_parameter(Name=name, Value=encoded_value, Type="String", Overwrite=overwrite)
self.log.info(f"Parameter '{name}' added/updated successfully with compression.")
return
except Exception as e:
self.log.critical(f"Failed to add/update compressed parameter '{name}': {e}")
raise
# Add or update the parameter normally if under 4KB
self.ssm_client.put_parameter(Name=name, Value=value, Type="String", Overwrite=overwrite)
self.log.info(f"Parameter '{name}' added/updated successfully.")
except Exception as e:
self.log.critical(f"Failed to add/update parameter '{name}': {e}")
raise
|
delete(name)
Delete a parameter from the AWS Parameter Store.
Parameters:
Name |
Type |
Description |
Default |
name |
str
|
The name of the parameter to delete.
|
required
|
Source code in src/sageworks/api/parameter_store.py
| def delete(self, name: str):
"""Delete a parameter from the AWS Parameter Store.
Args:
name (str): The name of the parameter to delete.
"""
try:
# Delete the parameter from Parameter Store
self.ssm_client.delete_parameter(Name=name)
self.log.info(f"Parameter '{name}' deleted successfully.")
except Exception as e:
self.log.error(f"Failed to delete parameter '{name}': {e}")
|
get(name, warn=True, decrypt=True)
Retrieve a parameter value from the AWS Parameter Store.
Parameters:
Name |
Type |
Description |
Default |
name |
str
|
The name of the parameter to retrieve.
|
required
|
warn |
bool
|
Whether to log a warning if the parameter is not found.
|
True
|
decrypt |
bool
|
Whether to decrypt secure string parameters.
|
True
|
Returns:
Type |
Description |
Union[str, list, dict, None]
|
Union[str, list, dict, None]: The value of the parameter or None if not found.
|
Source code in src/sageworks/api/parameter_store.py
| def get(self, name: str, warn: bool = True, decrypt: bool = True) -> Union[str, list, dict, None]:
"""Retrieve a parameter value from the AWS Parameter Store.
Args:
name (str): The name of the parameter to retrieve.
warn (bool): Whether to log a warning if the parameter is not found.
decrypt (bool): Whether to decrypt secure string parameters.
Returns:
Union[str, list, dict, None]: The value of the parameter or None if not found.
"""
try:
# Retrieve the parameter from Parameter Store
response = self.ssm_client.get_parameter(Name=name, WithDecryption=decrypt)
value = response["Parameter"]["Value"]
# Auto-detect and decompress if needed
if value.startswith("COMPRESSED:"):
# Base64 decode and decompress
self.log.important(f"Decompressing parameter '{name}'...")
compressed_value = base64.b64decode(value[len("COMPRESSED:") :])
value = zlib.decompress(compressed_value).decode("utf-8")
# Attempt to parse the value back to its original type
try:
parsed_value = json.loads(value)
return parsed_value
except (json.JSONDecodeError, TypeError):
# If parsing fails, return the value as is (assumed to be a simple string)
return value
except ClientError as e:
if e.response["Error"]["Code"] == "ParameterNotFound":
if warn:
self.log.warning(f"Parameter '{name}' not found")
else:
self.log.error(f"Failed to get parameter '{name}': {e}")
return None
|
list()
List all parameters in the AWS Parameter Store.
Returns:
Name | Type |
Description |
list |
list
|
A list of parameter names and details.
|
Source code in src/sageworks/api/parameter_store.py
| def list(self) -> list:
"""List all parameters in the AWS Parameter Store.
Returns:
list: A list of parameter names and details.
"""
try:
# Set up parameters for our search
params = {"MaxResults": 50}
# Initialize the list to collect parameter names
all_parameters = []
# Make the initial call to describe parameters
response = self.ssm_client.describe_parameters(**params)
# Aggregate the names from the initial response
all_parameters.extend(param["Name"] for param in response["Parameters"])
# Continue to paginate if there's a NextToken
while "NextToken" in response:
# Update the parameters with the NextToken for subsequent calls
params["NextToken"] = response["NextToken"]
response = self.ssm_client.describe_parameters(**params)
# Aggregate the names from the subsequent responses
all_parameters.extend(param["Name"] for param in response["Parameters"])
except Exception as e:
self.log.error(f"Failed to list parameters: {e}")
return []
# Return the aggregated list of parameter names
return all_parameters
|
Bypassing the 4k Limit
AWS Parameter Storage has a 4k limit on values, the SageWorks class bypasses this limit by detecting large values (strings, data, whatever) and compressing those on the fly. The decompressing is also handled automatically, so for larger data simply use the add()
and get()
methods and it will all just work.
Examples
These example show how to use the ParameterStore()
class to list, add, and get parameters from the AWS Parameter Store Service.
SageWorks REPL
If you'd like to experiment with listing, adding, and getting data with the ParameterStore()
class, you can spin up the SageWorks REPL, use the class and test out all the methods. Try it out! SageWorks REPL
Using SageWorks REPLparams = ParameterStore()
# List Parameters
params.list()
['/sageworks/abalone_info',
'/sageworks/my_data',
'/sageworks/test',
'/sageworks/pipelines/my_pipeline']
# Add Key
params.add("key", "value")
value = params.get("key")
# Add any data (lists, dictionaries, etc..)
my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
params.add("my_data", my_data)
# Retrieve data
return_value = params.get("my_data")
pprint(return_value)
{'key': 'value', 'list': [1, 2, 3], 'number': 4.2}
# Delete parameters
param_store.delete("my_data")
list()
not showing ALL parameters?
If you want access to ALL the parameters in the parameter store set prefix=None
and everything will show up.
params = ParameterStore(prefix=None)
params.list()
<all the keys>