Skip to content

Workbench Parameter Storage

Examples

Examples of using the Parameter Storage class are listed at the bottom of this page Examples.

ParameterStore: Manages Workbench parameters in AWS Systems Manager Parameter Store.

ParameterStore

ParameterStore: Manages Workbench parameters in AWS Systems Manager Parameter Store.

Common Usage 
params = ParameterStore()

# List Parameters
params.list()

['/workbench/abalone_info',
 '/workbench/my_data',
 '/workbench/test',
 '/workbench/pipelines/my_pipeline']

# Add Key
params.upsert("key", "value")
value = params.get("key")

# Add any data (lists, dictionaries, etc..)
my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
params.upsert("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/workbench/api/parameter_store.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class ParameterStore:
    """ParameterStore: Manages Workbench parameters in AWS Systems Manager Parameter Store.

    Common Usage:
        ```python
        params = ParameterStore()

        # List Parameters
        params.list()

        ['/workbench/abalone_info',
         '/workbench/my_data',
         '/workbench/test',
         '/workbench/pipelines/my_pipeline']

        # Add Key
        params.upsert("key", "value")
        value = params.get("key")

        # Add any data (lists, dictionaries, etc..)
        my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
        params.upsert("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("workbench")

        # Initialize a Workbench Session (to assume the Workbench 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, prefix: str = None) -> list:
        """List all parameters in the AWS Parameter Store, optionally filtering by a prefix.

        Args:
            prefix (str, optional): A prefix to filter the parameters by. Defaults to None.

        Returns:
            list: A list of parameter names and details.
        """
        try:
            # Set up parameters for the query
            params = {"MaxResults": 50}

            # If a prefix is provided, add the 'ParameterFilters' for optimization
            if prefix:
                params["ParameterFilters"] = [{"Key": "Name", "Option": "BeginsWith", "Values": [prefix]}]

            # 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 upsert(self, name: str, value, overwrite: bool = True):
        """Insert 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/workbench/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

                # Insert or update the compressed parameter in Parameter Store
                try:
                    # Insert 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

            # Insert 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/workbench/api/parameter_store.py 
48
49
50
51
52
53
54
55
56
def __init__(self):
    """ParameterStore Init Method"""
    self.log = logging.getLogger("workbench")

    # Initialize a Workbench Session (to assume the Workbench 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/workbench/api/parameter_store.py 
198
199
200
def __repr__(self):
    """Return a string representation of the ParameterStore object."""
    return "\n".join(self.list())

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/workbench/api/parameter_store.py 
185
186
187
188
189
190
191
192
193
194
195
196
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/workbench/api/parameter_store.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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(prefix=None)

List all parameters in the AWS Parameter Store, optionally filtering by a prefix.

Parameters:

Name Type Description Default
prefix str

A prefix to filter the parameters by. Defaults to None.

 None

Returns:

Name Type Description
list list

A list of parameter names and details.
Source code in src/workbench/api/parameter_store.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def list(self, prefix: str = None) -> list:
    """List all parameters in the AWS Parameter Store, optionally filtering by a prefix.

    Args:
        prefix (str, optional): A prefix to filter the parameters by. Defaults to None.

    Returns:
        list: A list of parameter names and details.
    """
    try:
        # Set up parameters for the query
        params = {"MaxResults": 50}

        # If a prefix is provided, add the 'ParameterFilters' for optimization
        if prefix:
            params["ParameterFilters"] = [{"Key": "Name", "Option": "BeginsWith", "Values": [prefix]}]

        # 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

upsert(name, value, overwrite=True)

Insert 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/workbench/api/parameter_store.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def upsert(self, name: str, value, overwrite: bool = True):
    """Insert 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/workbench/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

            # Insert or update the compressed parameter in Parameter Store
            try:
                # Insert 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

        # Insert 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

Bypassing the 4k Limit

AWS Parameter Storage has a 4k limit on values, the Workbench 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.

Workbench REPL

If you'd like to experiment with listing, adding, and getting data with the ParameterStore() class, you can spin up the Workbench REPL, use the class and test out all the methods. Try it out! Workbench REPL

Using Workbench REPL
params = ParameterStore()

# List Parameters
params.list()

['/workbench/abalone_info',
 '/workbench/my_data',
 '/workbench/test',
 '/workbench/pipelines/my_pipeline']

# Add Key
params.upsert("key", "value")
value = params.get("key")

# Add any data (lists, dictionaries, etc..)
my_data = {"key": "value", "number": 4.2, "list": [1,2,3]}
params.upsert("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>