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

@@ -29,7 +29,7 @@
 from aws_lambda_powertools.shared.functions import resolve_max_age
 from aws_lambda_powertools.utilities.parameters.types import TransformOptions
 
-from .exceptions import GetParameterError, TransformParameterError
+from .exceptions import GetParameterError, SetParameterError, TransformParameterError
 
 if TYPE_CHECKING:
     from mypy_boto3_appconfigdata import AppConfigDataClient
@@ -153,6 +153,13 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes, Dict[str, Any]]:
         """
         raise NotImplementedError()
 
+    @abstractmethod
+    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

@@ -6,6 +6,8 @@
 class GetParameterError(Exception):
     """When a provider raises an exception on parameter retrieval"""
 
-
 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

@@ -1,9 +1,10 @@
 """
 AWS Secrets Manager parameter retrieval and caching utility
 """
-
+from __future__ import annotations
 
 import os
+import json
 from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
 import boto3
@@ -16,6 +17,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):
@@ -116,6 +118,69 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         """
         raise NotImplementedError()
 
+    def set_secret(
+            self,
+            *, # force keyword arguments
+            name: str,
+            secret: Optional[str],
+            secret_binary: Optional[str],
+            idempotency_id: Optional[str],
+            version_stages: Optional[list[str]],
+            **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.
+        secret: str, optional
+            Specifies text data that you want to encrypt and store in this new version of the secret.
+        secret_binary: bytes, optional
+            Specifies binary data that you want to encrypt and store in this new version of the secret.
+        idempotency_id: str, optional
+            Idempotency token to use for the request to prevent the accidental
+            creation of duplicate versions if there are failures and retries
+            during the Lambda rotation function processing.
+        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
+        """
+
+        if isinstance(secret, dict):
+            secret = json.dumps(secret)
+
+        if not isinstance(secret_binary, bytes):
+            secret_binary = secret_binary.encode("utf-8")
+
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["SecretId"] = name
+        if secret:
+            sdk_options["SecretString"] = secret
+        if secret_binary:
+            sdk_options["SecretBinary"] = secret_binary
+        if version_stages:
+            sdk_options["VersionStages"] = version_stages
+        if idempotency_id:
+            sdk_options["ClientRequestToken"] = idempotency_id
+
+        try:
+            value = self.client.put_secret_value(**sdk_options)["VersionId"]
+        except Exception as exc:
+            raise SetParameterError(str(exc)) from exc
+
+        return value
+
 
 def get_secret(
     name: str,
@@ -183,3 +248,70 @@ def get_secret(
         force_fetch=force_fetch,
         **sdk_options,
     )
+
+
+def set_secret(
+    *, # force keyword arguments
+    name: str,
+    secret: Optional[Union[str, dict]] = None,
+    secret_binary: Optional[bytes] = None,
+    idempotency_id: 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
+    secret: str, optional
+        Secret value to set
+    secret_binary: bytes, optional
+        Secret binary value to set
+    idempotency_token: str, optional
+        Idempotency token to use for the request to prevent the accidental
+        creation of duplicate versions if there are failures and retries
+        during the Lambda rotation function processing.
+    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", secret="supers3cr3tllam@passw0rd")
+
+    **Sets a secret and includes an idempotency_id**
+
+        >>> from aws_lambda_powertools.utilities import parameters
+        >>>
+        >>> parameters.set_secret(name="my-secret", secret='{"password": "supers3cr3tllam@passw0rd"}', idempotency_id="f658cac0-98a5-41d9-b993-8a76a7799194")
+    """
+
+    # 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_secret(
+        name=name, secret=secret, secret_binary=secret_binary, idempotency_id=idempotency_id, version_stages=version_stages, **sdk_options
+    )

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -18,7 +18,7 @@
 from aws_lambda_powertools.shared.types import Literal
 
 from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider, transform_value
-from .exceptions import GetParameterError
+from .exceptions import GetParameterError, SetParameterError
 from .types import TransformOptions
 
 if TYPE_CHECKING:
@@ -188,6 +188,71 @@ def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str:
 
         return self.client.get_parameter(**sdk_options)["Parameter"]["Value"]
 
+    def set(
+        self,
+        path: str,
+        value: str,
+        *, # force keyword arguments
+        parameter_type: Optional[Literal["String", "StringList", "SecureString"]] = "String",
+        overwrite: bool = False,
+        tier: Optional[Literal["Standard", "Advanced", "Intelligent-Tiering"]] = "Standard",
+        description: Optional[str] = None,
+        kms_key_id: Optional[str] = None,
+        transform: Optional[str] = None,
+        **sdk_options,
+    ) -> int:
+        """
+        Retrieve a parameter value or return the cached value
+
+        Parameters
+        ----------
+        path: str
+            The fully qualified name includes the complete hierarchy of the parameter path and name.
+        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
+        transform: str, optional
+            Transforms the content from a JSON object ('json') or base64 binary string ('binary')
+        sdk_options: dict, optional
+            Dictionary of options that will be passed to the Parameter Store get_parameter API call
+
+        Raises
+        ------
+        SetParameterError
+            When the parameter provider fails to set a parameter value for
+            a given name.
+        TransformParameterError
+            When the parameter provider fails to transform a parameter value.
+        """
+
+        sdk_options["Name"] = path
+        sdk_options["Value"] = value
+        sdk_options["Type"] = parameter_type
+        sdk_options["Overwrite"] = overwrite
+        sdk_options["Tier"] = tier
+
+        if description:
+            sdk_options["Description"] = description
+        if kms_key_id:
+            sdk_options["KeyId"] = kms_key_id
+
+        try:
+            value = self.client.put_parameter(**sdk_options)["Version"]
+        except Exception as exc:
+            raise SetParameterError(str(exc)) from exc
+
+        if transform:
+            value = transform_value(key=path, value=value, transform=transform, raise_on_transform_error=True)
+
+        return value
+
     def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = False, **sdk_options) -> Dict[str, str]:
         """
         Retrieve multiple parameter values from AWS Systems Manager Parameter Store
@@ -702,6 +767,85 @@ def get_parameters(
     )
 
 
+def set_parameter(
+    path: str,
+    value: str,
+    *, # force keyword arguments
+    parameter_type: Optional[Literal["String", "StringList", "SecureString"]] = "String",
+    overwrite: bool = False,
+    tier: Optional[Literal["Standard", "Advanced", "Intelligent-Tiering"]] = "Standard",
+    description: Optional[str] = None,
+    kms_key_id: Optional[str] = None,
+    transform: 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.
+    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
+    transform: str, optional
+        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
+    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.
+    TransformParameterError
+        When the parameter provider fails to transform a parameter value.
+
+    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,
+        transform=transform,
+        **sdk_options
+    )
+
+
 @overload
 def get_parameters_by_name(
     parameters: Dict[str, Dict],