Skip to content

Workbench Parameter Storage

The Parameter Store is a great place to publish data (strings, lists, dictionaries, etc). The service is provided on all AWS accounts and allows ML pipelines to create, store, and read data.

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.

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 ParameterStore
from workbench.api import ParameterStore
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>

ParameterStore: Manages Workbench parameters in a Cloud Based Parameter Store.

ParameterStore

Bases: AWSParameterStore

ParameterStore: Manages Workbench parameters in a Cloud Based 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 
10
11
12
13
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
class ParameterStore(AWSParameterStore):
    """ParameterStore: Manages Workbench parameters in a Cloud Based 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 the SuperClass
        super().__init__()

    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.
        """
        return super().list(prefix=prefix)

    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.
        """
        return super().get(name=name, warn=warn, decrypt=decrypt)

    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)
        """
        super().upsert(name=name, value=value, overwrite=overwrite)

    def delete(self, name: str):
        """Delete a parameter from the AWS Parameter Store.

        Args:
            name (str): The name of the parameter to delete.
        """
        super().delete(name=name)

    def __repr__(self):
        """Return a string representation of the ParameterStore object."""
        return super().__repr__()

__init__()

ParameterStore Init Method

Source code in src/workbench/api/parameter_store.py 
44
45
46
47
48
49
def __init__(self):
    """ParameterStore Init Method"""
    self.log = logging.getLogger("workbench")

    # Initialize the SuperClass
    super().__init__()

__repr__()

Return a string representation of the ParameterStore object.

Source code in src/workbench/api/parameter_store.py 
93
94
95
def __repr__(self):
    """Return a string representation of the ParameterStore object."""
    return super().__repr__()

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 
85
86
87
88
89
90
91
def delete(self, name: str):
    """Delete a parameter from the AWS Parameter Store.

    Args:
        name (str): The name of the parameter to delete.
    """
    super().delete(name=name)

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 
62
63
64
65
66
67
68
69
70
71
72
73
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.
    """
    return super().get(name=name, warn=warn, decrypt=decrypt)

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 
51
52
53
54
55
56
57
58
59
60
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.
    """
    return super().list(prefix=prefix)

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 
75
76
77
78
79
80
81
82
83
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)
    """
    super().upsert(name=name, value=value, overwrite=overwrite)