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, get_parameters, get_parameters_by_name, set_parameter
 
 __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,9 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes, Dict[str, Any]]:
         """
         raise NotImplementedError()
 
+    def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs):
+        raise NotImplementedError()
+
     def get_multiple(
         self,
         path: str,

aws_lambda_powertools/utilities/parameters/exceptions.py

@@ -9,3 +9,11 @@ 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 SSM parameter"""
+
+
+class SetSecretError(Exception):
+    """When a provider raises an exception on setting an AWS Secret"""

aws_lambda_powertools/utilities/parameters/secrets.py

@@ -2,21 +2,27 @@
 AWS Secrets Manager parameter retrieval and caching utility
 """
 
+from __future__ import annotations
+
+import json
+import logging
 import os
 from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Union, overload
 
 import boto3
 from botocore.config import Config
 
-from aws_lambda_powertools.utilities.parameters.types import TransformOptions
+from aws_lambda_powertools.utilities.parameters.types import SetSecretResponse, TransformOptions
 
 if TYPE_CHECKING:
     from mypy_boto3_secretsmanager import SecretsManagerClient
 
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import resolve_max_age
+from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
+from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError
 
-from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
+logger = logging.getLogger(__name__)
 
 
 class SecretsProvider(BaseProvider):
@@ -117,6 +123,135 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         """
         raise NotImplementedError()
 
+    def _create_secret(self, name: str, **sdk_options):
+        """
+        Create a secret with the given name.
+
+        Parameters:
+        ----------
+        name: str
+            The name of the secret.
+        **sdk_options:
+            Additional options to be passed to the create_secret method.
+
+        Raises:
+            SetSecretError: If there is an error setting the secret.
+        """
+        try:
+            sdk_options["Name"] = name
+            return self.client.create_secret(**sdk_options)
+        except Exception as exc:
+            raise SetSecretError(f"Error setting secret - {str(exc)}") from exc
+
+    def _update_secret(self, name: str, **sdk_options):
+        """
+        Update a secret with the given name.
+
+        Parameters:
+        ----------
+        name: str
+            The name of the secret.
+        **sdk_options:
+            Additional options to be passed to the create_secret method.
+        """
+        sdk_options["SecretId"] = name
+        return self.client.put_secret_value(**sdk_options)
+
+    def set(
+        self,
+        name: str,
+        value: Union[str, dict, bytes],
+        *,  # force keyword arguments
+        client_request_token: Optional[str] = None,
+        **sdk_options,
+    ) -> SetSecretResponse:
+        """
+        Modify the details of a secret or create a new secret if it doesn't already exist.
+        It includes metadata and the secret value.
+
+        We aim to minimize API calls by assuming that the secret already exists and needs updating.
+        If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding:
+
+
+                          ┌────────────────────────┐      ┌─────────────────┐
+                ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐
+                │         └────────────────────────┘      └─────────────────┘     │
+                │                                                                 │
+                │                                                                 │
+                │                                                                 ▼
+        ┌─────────────────┐                                              ┌─────────────────────┐
+        │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │
+        └─────────────────┘                                              └─────────────────────┘
+
+        Parameters
+        ----------
+        name: str
+            The ARN or name of the secret to add a new version to or create a new one.
+        value: str, dict 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
+            auto-populated if not provided.
+        sdk_options: dict, optional
+            Dictionary of options that will be passed to the Secrets Manager update_secret API call
+
+        Raises
+        ------
+        SetSecretError
+            When attempting to update or create a secret fails.
+
+        Returns:
+        -------
+        SetSecretResponse:
+            The dict returned by boto3.
+
+        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
+            >>> import uuid
+            >>>
+            >>> parameters.set_secret(
+                    name="my-secret",
+                    value='{"password": "supers3cr3tllam@passw0rd"}',
+                    client_request_token=str(uuid.uuid4())
+                )
+
+        URLs:
+        -------
+            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html
+            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html
+        """
+
+        if isinstance(value, dict):
+            value = json.dumps(value)
+
+        if isinstance(value, bytes):
+            sdk_options["SecretBinary"] = value
+        else:
+            sdk_options["SecretString"] = value
+
+        if client_request_token:
+            sdk_options["ClientRequestToken"] = client_request_token
+
+        try:
+            logger.debug(f"Attempting to update secret {name}")
+            return self._update_secret(name=name, **sdk_options)
+        except self.client.exceptions.ResourceNotFoundException:
+            logger.debug(f"Secret {name} doesn't exist, creating a new one")
+            return self._create_secret(name=name, **sdk_options)
+        except Exception as exc:
+            raise SetSecretError(f"Error setting secret - {str(exc)}") from exc
+
 
 @overload
 def get_secret(
@@ -224,3 +359,88 @@ 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,
+    **sdk_options,
+) -> str:
+    """
+    Modify the details of a secret or create a new secret if it doesn't already exist.
+    It includes metadata and the secret value.
+
+    We aim to minimize API calls by assuming that the secret already exists and needs updating.
+    If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding:
+
+
+                      ┌────────────────────────┐      ┌─────────────────┐
+            ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐
+            │         └────────────────────────┘      └─────────────────┘     │
+            │                                                                 │
+            │                                                                 │
+            │                                                                 ▼
+    ┌─────────────────┐                                              ┌─────────────────────┐
+    │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │
+    └─────────────────┘                                              └─────────────────────┘
+
+    Parameters
+    ----------
+    name: str
+        The ARN or name of the secret to add a new version to or create a new one.
+    value: str, dict 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
+        auto-populated if not provided.
+    sdk_options: dict, optional
+        Dictionary of options that will be passed to the Secrets Manager update_secret API call
+
+    Raises
+    ------
+    SetSecretError
+        When attempting to update or create a secret fails.
+
+    Returns:
+    -------
+    SetSecretResponse:
+        The dict returned by boto3.
+
+    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"
+            )
+
+    URLs:
+    -------
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html
+        https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html
+    """
+
+    # 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,
+        **sdk_options,
+    )