feat(parameters): add feature for creating and updating Parameters and Secrets

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -8,8 +8,8 @@
 from .base import BaseProvider, clear_caches
 from .dynamodb import DynamoDBProvider
 from .exceptions import GetParameterError, TransformParameterError
-from .secrets import SecretsProvider, get_secret
-from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name
+from .secrets import SecretsProvider, get_secret, set_secret
+from .ssm import SSMProvider, get_parameter, set_parameter, get_parameters, get_parameters_by_name
 
 __all__ = [
     "AppConfigProvider",
@@ -21,8 +21,10 @@
     "TransformParameterError",
     "get_app_config",
     "get_parameter",
+    "set_parameter",
     "get_parameters",
     "get_parameters_by_name",
     "get_secret",
+    "set_secret",
     "clear_caches",
 ]

aws_lambda_powertools/utilities/parameters/base.py

@@ -154,6 +154,12 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes, Dict[str, Any]]:
         """
         raise NotImplementedError()
 
+    def _set(self, name: str, **sdk_options) -> Union[str, bytes]:
+        """
+        Sets a parameter value from the underlying parameter store
+        """
+        raise NotImplementedError()
+
     def get_multiple(
         self,
         path: str,

aws_lambda_powertools/utilities/parameters/exceptions.py

@@ -9,3 +9,7 @@ class GetParameterError(Exception):
 
 class TransformParameterError(Exception):
     """When a provider fails to transform a parameter value"""
+
+
+class SetParameterError(Exception):
+    """When a provider raises an exception on setting a parameter"""

aws_lambda_powertools/utilities/parameters/secrets.py

@@ -2,11 +2,15 @@
 AWS Secrets Manager parameter retrieval and caching utility
 """
 
+from __future__ import annotations
+
+import json
 import os
 from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
 import boto3
 from botocore.config import Config
+from botocore.exceptions import ClientError
 
 if TYPE_CHECKING:
     from mypy_boto3_secretsmanager import SecretsManagerClient
@@ -15,6 +19,7 @@
 from aws_lambda_powertools.shared.functions import resolve_max_age
 
 from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
+from .exceptions import SetParameterError
 
 
 class SecretsProvider(BaseProvider):
@@ -115,6 +120,65 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         """
         raise NotImplementedError()
 
+    def _set(
+        self,
+        name: str,
+        value: Union[str, dict, bytes],
+        *,  # force keyword arguments
+        client_request_token: Optional[str] = None,
+        version_stages: Optional[list[str]] = None,
+        **sdk_options,
+    ) -> str:
+        """
+        Modifies the details of a secret, including metadata and the secret value.
+
+        Parameters
+        ----------
+        name: str
+            The ARN or name of the secret to add a new version to.
+        value: str or bytes
+            Specifies text data that you want to encrypt and store in this new version of the secret.
+        client_request_token: str, optional
+            This value helps ensure idempotency. Recommended that you generate
+            a UUID-type value to ensure uniqueness within the specified secret.
+            This value becomes the VersionId of the new version. This field is
+            autopopulated if not provided.
+        version_stages: list[str], optional
+            Specifies a list of staging labels that are attached to this version of the secret.
+        sdk_options: dict, optional
+            Dictionary of options that will be passed to the Secrets Manager update_secret API call
+
+        Returns:
+        -------
+            Version ID of the newly created version of the secret.
+
+        URLs:
+        -------
+            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html
+        """
+
+        sdk_options["SecretId"] = name
+
+        if isinstance(value, dict):
+            value = json.dumps(value)
+
+        if isinstance(value, bytes):
+            sdk_options["SecretBinary"] = value
+        else:
+            sdk_options["SecretString"] = value
+
+        if version_stages:
+            sdk_options["VersionStages"] = version_stages
+        if client_request_token:
+            sdk_options["ClientRequestToken"] = client_request_token
+
+        try:
+            value = self.client.put_secret_value(**sdk_options)
+            return value["VersionId"]
+        except ClientError as exc:
+            if exc.response["Error"]["Code"] != "ResourceNotFoundException":
+                raise SetParameterError(str(exc)) from exc
+
 
 def get_secret(
     name: str,
@@ -182,3 +246,76 @@ def get_secret(
         force_fetch=force_fetch,
         **sdk_options,
     )
+
+
+def set_secret(
+    name: str,
+    value: Union[str, bytes],
+    *,  # force keyword arguments
+    client_request_token: Optional[str] = None,
+    version_stages: Optional[list[str]] = None,
+    **sdk_options,
+) -> str:
+    """
+    Retrieve a parameter value from AWS Secrets Manager
+
+    Parameters
+    ----------
+    name: str
+        Name of the parameter
+    value: str or bytes
+        Secret value to set
+    client_request_token: str, optional
+        This value helps ensure idempotency. Recommended that you generate
+        a UUID-type value to ensure uniqueness within the specified secret.
+        This value becomes the VersionId of the new version. This field is
+        autopopulated if not provided.
+    version_stages: list[str], optional
+        A list of staging labels that are attached to this version of the secret.
+    sdk_options: dict, optional
+        Dictionary of options that will be passed to the get_secret_value call
+
+    Raises
+    ------
+    SetParameterError
+        When the secrets provider fails to set a secret value or secret binary for
+        a given name.
+
+    Returns:
+    -------
+        Version ID of the newly created version of the secret.
+
+    URLs:
+    -------
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html
+
+    Example
+    -------
+    **Sets a secret***
+
+        >>> from aws_lambda_powertools.utilities import parameters
+        >>>
+        >>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd")
+
+    **Sets a secret and includes an client_request_token**
+
+        >>> from aws_lambda_powertools.utilities import parameters
+        >>>
+        >>> parameters.set_secret(
+                name="my-secret",
+                value='{"password": "supers3cr3tllam@passw0rd"}',
+                client_request_token="61f2af5f-5f75-44b1-a29f-0cc37af55b11"
+            )
+    """
+
+    # Only create the provider if this function is called at least once
+    if "secrets" not in DEFAULT_PROVIDERS:
+        DEFAULT_PROVIDERS["secrets"] = SecretsProvider()
+
+    return DEFAULT_PROVIDERS["secrets"]._set(
+        name=name,
+        value=value,
+        client_request_token=client_request_token,
+        version_stages=version_stages,
+        **sdk_options,
+    )

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -26,6 +26,9 @@
     from mypy_boto3_ssm import SSMClient
     from mypy_boto3_ssm.type_defs import GetParametersResultTypeDef
 
+SSM_PARAMETER_TYPES = Literal["String", "StringList", "SecureString"]
+SSM_PARAMETER_TIER = Literal["Standard", "Advanced", "Intelligent-Tiering"]
+
 
 class SSMProvider(BaseProvider):
     """
@@ -811,6 +814,81 @@ def get_parameters(
     )
 
 
+def set_parameter(
+    path: str,
+    value: str,
+    *,  # force keyword arguments
+    parameter_type: SSM_PARAMETER_TYPES = "String",
+    overwrite: bool = False,
+    tier: SSM_PARAMETER_TIER = "Standard",
+    description: Optional[str] = None,
+    kms_key_id: Optional[str] = None,
+    **sdk_options,
+) -> int:
+    """
+    Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store
+
+    Parameters
+    ----------
+    path: str
+        The fully qualified name includes the complete hierarchy of the parameter path and name.
+    value: str
+        The parameter value
+    parameter_type: str, optional
+        Type of the parameter.  Allowed values are String, StringList, and SecureString
+    overwrite: bool, optional
+        If the parameter value should be overwritten, False by default
+    tier: str, optional
+        The parameter tier to use.  Allowed values are Standard, Advanced, and Intelligent-Tiering
+    description: str, optional
+        The description of the parameter
+    kms_key_id: str, optional
+        The KMS key id to use to encrypt the parameter
+    sdk_options: dict, optional
+        Dictionary of options that will be passed to the Parameter Store get_parameter API call
+
+    Returns:
+    --------
+        The version (integer) of the parameter that was set
+
+    Raises
+    ------
+    SetParameterError
+        When the parameter provider fails to retrieve a parameter value for
+        a given name.
+
+    URLs:
+    -------
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html
+
+    Example
+    -------
+    **Sets a parameter value from Systems Manager Parameter Store**
+
+        >>> from aws_lambda_powertools.utilities import parameters
+        >>>
+        >>> response = parameters.set_parameter(path="/my/example/parameter", value="More Powertools")
+        >>>
+        >>> print(response)
+        123
+    """
+
+    # Only create the provider if this function is called at least once
+    if "ssm" not in DEFAULT_PROVIDERS:
+        DEFAULT_PROVIDERS["ssm"] = SSMProvider()
+
+    return DEFAULT_PROVIDERS["ssm"]._set(
+        path,
+        value,
+        parameter_type=parameter_type,
+        overwrite=overwrite,
+        tier=tier,
+        description=description,
+        kms_key_id=kms_key_id,
+        **sdk_options,
+    )
+
+
 @overload
 def get_parameters_by_name(
     parameters: Dict[str, Dict],

docs/utilities/parameters.md

@@ -32,8 +32,10 @@ This utility requires additional permissions to work as expected.
 | SSM       | **`get_parameter`**, **`SSMProvider.get`**                             | **`ssm:GetParameter`**                                                               |
 | SSM       | **`get_parameters`**, **`SSMProvider.get_multiple`**                   | **`ssm:GetParametersByPath`**                                                        |
 | SSM       | **`get_parameters_by_name`**, **`SSMProvider.get_parameters_by_name`** | **`ssm:GetParameter`** and **`ssm:GetParameters`**                                   |
+| SSM       | **`set_parameter`**                                                    | **`ssm:PutParameter`**                                                               |
 | SSM       | If using **`decrypt=True`**                                            | You must add an additional permission **`kms:Decrypt`**                              |
 | Secrets   | **`get_secret`**, **`SecretsProvider.get`**                            | **`secretsmanager:GetSecretValue`**                                                  |
+| Secrets   | **`set_secret`**, **`SecretsProvider.get`**                            | **`secretsmanager:PutSecretValue`** and or **`secretsmanager:CreateSecret`**         |
 | DynamoDB  | **`DynamoDBProvider.get`**                                             | **`dynamodb:GetItem`**                                                               |
 | DynamoDB  | **`DynamoDBProvider.get_multiple`**                                    | **`dynamodb:Query`**                                                                 |
 | AppConfig | **`get_app_config`**, **`AppConfigProvider.get_app_config`**           | **`appconfig:GetLatestConfiguration`** and **`appconfig:StartConfigurationSession`** |
@@ -84,6 +86,23 @@ For multiple parameters, you can use either:
     --8<-- "examples/parameters/src/get_parameter_by_name_error_handling.py"
     ```
 
+### Setting parameters
+
+You can set a parameter using the `set_parameter` high-level function. This will create a new parameter if it doesn't exist.
+
+=== "getting_started_set_single_ssm_parameter.py"
+    ```python hl_lines="8"
+    --8<-- "examples/parameters/src/getting_started_set_single_ssm_parameter.py"
+    ```
+
+=== "getting_started_set_ssm_parameter_overwrite.py"
+    There are occasions where sometimes you are setting a parameter and then you may need to update that parameter later on. In this case, you can use the `overwrite` parameter to overwrite the parameter value if it already exists. If you do not set this parameter, then the parameter will not be overwritten and an exception will be raised.
+
+    ```python hl_lines="8"
+    --8<-- "examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py"
+    ```
+
+
 ### Fetching secrets
 
 You can fetch secrets stored in Secrets Manager using `get_secret`.
@@ -93,6 +112,16 @@ You can fetch secrets stored in Secrets Manager using `get_secret`.
     --8<-- "examples/parameters/src/getting_started_secret.py"
     ```
 
+### Setting secrets
+
+You can set secrets stored in Secrets Manager using `set_secret`.
+
+=== "getting_started_secret.py"
+    ```python hl_lines="5 15"
+    --8<-- "examples/parameters/src/getting_started_setting_secret.py"
+    ```
+
+
 ### Fetching app configurations
 
 You can fetch application configurations in AWS AppConfig using `get_app_config`.

examples/parameters/src/getting_started_set_single_ssm_parameter.py

@@ -0,0 +1,12 @@
+from aws_lambda_powertools.utilities import parameters
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+def lambda_handler(event: dict, context: LambdaContext) -> dict:
+    try:
+        # Set a single parameter, returns the version ID of the parameter
+        parameter_version = parameters.set_parameter(path="/mySuper/Parameter", value="PowerToolsIsAwesome")  # type: ignore[assignment] # noqa: E501
+
+        return {"mySuperParameterVersion": parameter_version, "statusCode": 200}
+    except parameters.exceptions.SetParameterError as error:
+        return {"comments": None, "message": str(error), "statusCode": 400}

examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py

@@ -0,0 +1,13 @@
+from aws_lambda_powertools.utilities import parameters
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+def lambda_handler(event: dict, context: LambdaContext) -> dict:
+    try:
+        # Set a single parameter, but overwrite if it already exists
+        # by default, overwrite is False, explicitly set it to True
+        updating_parameter = parameters.set_parameter(path="/mySuper/Parameter", value="PowerToolsIsAwesome", overwrite=True)  # type: ignore[assignment] # noqa: E501
+
+        return {"mySuperParameterVersion": updating_parameter, "statusCode": 200}
+    except parameters.exceptions.SetParameterError as error:
+        return {"comments": None, "message": str(error), "statusCode": 400}