feat(parameters): add support for retrieving batch of secrets

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -6,7 +6,7 @@
 from .base import BaseProvider, clear_caches
 from .dynamodb import DynamoDBProvider
 from .exceptions import GetParameterError, TransformParameterError
-from .secrets import SecretsProvider, get_secret, set_secret
+from .secrets import SecretsProvider, get_secret, get_secrets_by_name, set_secret
 from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name, set_parameter
 
 __all__ = [
@@ -23,6 +23,7 @@
     "get_parameters",
     "get_parameters_by_name",
     "get_secret",
+    "get_secrets_by_name",
     "set_secret",
     "clear_caches",
 ]

aws_lambda_powertools/utilities/parameters/exceptions.py

@@ -7,6 +7,10 @@ class GetParameterError(Exception):
     """When a provider raises an exception on parameter retrieval"""
 
 
+class GetSecretError(Exception):
+    """When a provider raises an exception on secret retrieval"""
+
+
 class TransformParameterError(Exception):
     """When a provider fails to transform a parameter value"""
 

aws_lambda_powertools/utilities/parameters/secrets.py

@@ -8,16 +8,20 @@
 import logging
 import os
 import warnings
-from typing import TYPE_CHECKING, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal, overload
 
 import boto3
 
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import resolve_max_age
 from aws_lambda_powertools.shared.json_encoder import Encoder
-from aws_lambda_powertools.utilities.parameters.base import BaseProvider
+from aws_lambda_powertools.utilities.parameters.base import BaseProvider, transform_value
 from aws_lambda_powertools.utilities.parameters.constants import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS
-from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError
+from aws_lambda_powertools.utilities.parameters.exceptions import (
+    GetSecretError,
+    SetSecretError,
+    TransformParameterError,
+)
 from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning
 
 if TYPE_CHECKING:
@@ -126,11 +130,159 @@ def _get(self, name: str, **sdk_options) -> str | bytes:
 
         return secret_value["SecretBinary"]
 
-    def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]:
+    def _get_multiple(self, names: list[str], **sdk_options) -> dict[str, Any]:  # type: ignore[override]
         """
-        Retrieving multiple parameter values is not supported with AWS Secrets Manager
+        Retrieve multiple secrets using AWS Secrets Manager batch_get_secret_value API
+
+        Parameters
+        ----------
+        names: list[str]
+            List of secret names to retrieve
+        sdk_options: dict, optional
+            Additional options passed to batch_get_secret_value API call
+
+        Returns
+        -------
+        dict[str, str]
+            Dictionary mapping secret names to their values
+
+        Raises
+        ------
+        GetParameterError
+            When the parameter provider fails to retrieve secrets
         """
-        raise NotImplementedError()
+
+        # Merge filters: combine names with any additional filters from sdk_options
+        filters = sdk_options.get("Filters", [])
+        name_filter = {"Key": "name", "Values": names}
+
+        # Add name filter to existing filters
+        filters.append(name_filter)
+        sdk_options["Filters"] = filters
+
+        # Remove SecretIdList if present to avoid conflicts
+        sdk_options.pop("SecretIdList", None)
+
+        secrets: dict[str, Any] = {}
+        next_token = None
+
+        # Handle pagination automatically
+        while True:
+            if next_token:
+                sdk_options["NextToken"] = next_token
+            elif "NextToken" in sdk_options:
+                # Remove NextToken from first call if it was passed
+                sdk_options.pop("NextToken")  # pragma: no cover
+
+            try:
+                response = self.client.batch_get_secret_value(**sdk_options)
+            except Exception as exc:
+                raise GetSecretError(f"Failed to retrieve secrets: {str(exc)}") from exc
+
+            # Process successful secrets
+            for secret in response.get("SecretValues", []):
+                secret_name = secret["Name"]
+
+                # Extract secret value (SecretString or SecretBinary)
+                if "SecretString" in secret:
+                    secrets[secret_name] = secret["SecretString"]
+                elif "SecretBinary" in secret:
+                    secrets[secret_name] = secret["SecretBinary"]
+
+            # Check if there are more results
+            next_token = response.get("NextToken")
+            if not next_token:
+                break
+
+        # If no secrets were found, raise an error
+        if not secrets:
+            raise GetSecretError(f"No secrets found matching the provided names: {names}")
+
+        return secrets
+
+    def get_multiple(  # type: ignore[override]
+        self,
+        names: list[str],
+        max_age: int | None = None,
+        transform: TransformOptions = None,
+        raise_on_transform_error: bool = False,
+        force_fetch: bool = False,
+        **sdk_options,
+    ) -> dict[str, Any]:
+        """
+        Retrieve multiple secrets by name from AWS Secrets Manager
+
+        Parameters
+        ----------
+        names: list[str]
+            List of secret names to retrieve
+        max_age: int, optional
+            Maximum age of the cached value
+        transform: str, optional
+            Optional transformation of the parameter value. Supported values
+            are "json" for JSON strings and "binary" for base 64 encoded values.
+        raise_on_transform_error: bool, optional
+            Raises an exception if any transform fails, otherwise this will
+            return a None value for each transform that failed
+        force_fetch: bool, optional
+            Force update even before a cached item has expired, defaults to False
+        sdk_options: dict, optional
+            Arguments that will be passed directly to the underlying API call
+
+        Returns
+        -------
+        dict[str, str | bytes | dict]
+            Dictionary mapping secret names to their values
+
+        Raises
+        ------
+        GetParameterError
+            When the parameter provider fails to retrieve secrets
+        TransformParameterError
+            When the parameter provider fails to transform a secret value
+        """
+        if not names:
+            raise GetSecretError("You must provide at least one secret name")
+
+        # Create a unique cache key for this batch of secrets
+        # Use sorted names to ensure consistent caching regardless of order
+        cache_key_name = "|".join(sorted(names))
+        key = self._build_cache_key(name=cache_key_name, transform=transform, is_nested=True)
+
+        # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
+        if not force_fetch and self.has_not_expired_in_cache(key):
+            cached_values = self.fetch_from_cache(key)
+            # Return only the requested secrets from cache (in case cache has more)
+            return {name: cached_values[name] for name in names if name in cached_values}
+
+        try:
+            values = self._get_multiple(names, **sdk_options)
+        except Exception as exc:
+            raise GetSecretError(str(exc)) from exc
+
+        if transform:
+            # Transform each secret value
+            transformed_values = {}
+            for name, value in values.items():
+                try:
+                    transformed_values[name] = transform_value(
+                        key=name,
+                        value=value,
+                        transform=transform,
+                        raise_on_transform_error=raise_on_transform_error,
+                    )
+                except TransformParameterError:
+                    if raise_on_transform_error:
+                        raise
+                    transformed_values[name] = None  # pragma: no cover
+            values = transformed_values
+
+        # Cache the results
+        self.add_to_cache(key=key, value=values, max_age=max_age)
+
+        return values
 
     def _create_secret(self, name: str, **sdk_options) -> CreateSecretResponseTypeDef:
         """
@@ -369,6 +521,85 @@ def get_secret(
     )
 
 
+def get_secrets_by_name(
+    names: list[str],
+    transform: TransformOptions = None,
+    force_fetch: bool = False,
+    max_age: int | None = None,
+    **sdk_options,
+) -> dict[str, str | bytes | dict]:
+    """
+    Retrieve multiple secrets by name from AWS Secrets Manager
+
+    Parameters
+    ----------
+    names: list[str]
+        List of secret names to retrieve
+    transform: str, optional
+        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
+    force_fetch: bool, optional
+        Force update even before a cached item has expired, defaults to False
+    max_age: int, optional
+        Maximum age of the cached value
+    sdk_options: dict, optional
+        Dictionary of options that will be passed to the batch_get_secret_value call
+
+    Raises
+    ------
+    GetParameterError
+        When the parameter provider fails to retrieve secrets
+    TransformParameterError
+        When the parameter provider fails to transform a secret value
+
+    Returns
+    -------
+    dict[str, str | bytes | dict]
+        Dictionary mapping secret names to their values
+
+    Example
+    -------
+    **Retrieves multiple secrets**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name
+        >>>
+        >>> secrets = get_secrets_by_name(["db-password", "api-key", "jwt-secret"])
+        >>> print(secrets["db-password"])
+
+    **Retrieves multiple secrets with JSON transformation**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name
+        >>>
+        >>> secrets = get_secrets_by_name(["config", "settings"], transform="json")
+        >>> print(secrets["config"]["database_url"])
+
+    **Retrieves multiple secrets with additional filters**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name
+        >>>
+        >>> secrets = get_secrets_by_name(
+        ...     names=["app-secret"],
+        ...     Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}]
+        ... )
+    """
+    if not names:
+        raise GetSecretError("You must provide at least one secret name")
+
+    # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
+    # 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"].get_multiple(
+        names=names,
+        max_age=max_age,
+        transform=transform,
+        force_fetch=force_fetch,
+        **sdk_options,
+    )
+
+
 def set_secret(
     name: str,
     value: str | bytes,

docs/utilities/parameters.md

@@ -36,6 +36,7 @@ This utility requires additional permissions to work as expected.
 | SSM       | If using **`decrypt=True`**                                            | You must add an additional permission **`kms:Decrypt`**                                         |
 | Secrets   | **`get_secret`**, **`SecretsProvider.get`**                            | **`secretsmanager:GetSecretValue`**                                                             |
 | Secrets   | **`set_secret`**, **`SecretsProvider.set`**                            | **`secretsmanager:PutSecretValue`** and **`secretsmanager:CreateSecret`** (if creating secrets) |
+| Secrets   | **`get_secrets_by_name`**, **`SecretsProvider.get_multiple`**          | **`secretsmanager:BatchGetSecretValue`**, **`secretsmanager:GetSecretValue`**  and **`secretsmanager:ListSecrets`** |
 | DynamoDB  | **`DynamoDBProvider.get`**                                             | **`dynamodb:GetItem`**                                                                          |
 | DynamoDB  | **`DynamoDBProvider.get_multiple`**                                    | **`dynamodb:Query`**                                                                            |
 | AppConfig | **`get_app_config`**, **`AppConfigProvider.get_app_config`**           | **`appconfig:GetLatestConfiguration`** and **`appconfig:StartConfigurationSession`**            |
@@ -111,6 +112,30 @@ You can fetch secrets stored in Secrets Manager using `get_secret`.
     --8<-- "examples/parameters/src/getting_started_secret.py"
     ```
 
+### Fetching multiple secrets
+
+You can fetch multiple secrets from Secrets Manager in a single API call using `get_secrets_by_name`. This reduces the number of API calls and improves performance when you need to retrieve several secrets at once.
+
+???+ info "Batch retrieval benefits"
+    - **Performance**: Retrieve up to 20 secrets in one API call
+    - **Cost optimization**: Fewer API calls reduce AWS costs
+    - **Error resilience**: Partial failures don't break the entire operation
+    - **Advanced filtering**: Use additional filters beyond secret names
+
+=== "getting_started_batch_secrets.py"
+    ```python hl_lines="1 7"
+    --8<-- "examples/parameters/src/getting_started_batch_secrets.py"
+    ```
+
+#### Advanced filtering
+
+You can combine secret name filtering with additional AWS Secrets Manager filters for more precise results:
+
+=== "batch_secrets_with_filters.py"
+    ```python hl_lines="2 10-16"
+    --8<-- "examples/parameters/src/batch_secrets_with_filters.py"
+    ```
+
 ### Setting secrets
 
 You can set secrets stored in Secrets Manager using `set_secret`.
@@ -251,6 +276,11 @@ You can create `SecureString` parameters, which are parameters that have a plain
     --8<-- "examples/parameters/src/builtin_provider_secret.py"
     ```
 
+=== "batch_secrets_provider.py"
+    ```python hl_lines="2-12"
+    --8<-- "examples/parameters/src/batch_secrets_provider.py"
+    ```
+
 #### DynamoDBProvider
 
 The DynamoDB Provider does not have any high-level functions, as it needs to know the name of the DynamoDB table containing the parameters.
@@ -445,7 +475,9 @@ Here is the mapping between this utility's functions and methods and the underly
 | SSM Parameter Store | `SSMProvider.get`               | `ssm`            | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter){target="_blank"}                                                                                                                                                                                                                                              |
 | SSM Parameter Store | `SSMProvider.get_multiple`      | `ssm`            | [get_parameters_by_path](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameters_by_path){target="_blank"}                                                                                                                                                                                                                            |
 | Secrets Manager     | `get_secret`                    | `secretsmanager` | [get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.get_secret_value){target="_blank"}                                                                                                                                                                                                                  |
+| Secrets Manager     | `get_secrets_by_name`           | `secretsmanager` | [batch_get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.batch_get_secret_value){target="_blank"}                                                                                                                                                                                                      |
 | Secrets Manager     | `SecretsProvider.get`           | `secretsmanager` | [get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.get_secret_value){target="_blank"}                                                                                                                                                                                                                  |
+| Secrets Manager     | `SecretsProvider.get_multiple`  | `secretsmanager` | [batch_get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.batch_get_secret_value){target="_blank"}                                                                                                                                                                                                      |
 | DynamoDB            | `DynamoDBProvider.get`          | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table){target="_blank"})    |
 | DynamoDB            | `DynamoDBProvider.get_multiple` | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table){target="_blank"})    |
 | App Config          | `get_app_config`                | `appconfigdata`  | [start_configuration_session](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata.html#AppConfigData.Client.start_configuration_session){target="_blank"} and [get_latest_configuration](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata.html#AppConfigData.Client.get_latest_configuration){target="_blank"} |

examples/parameters/src/batch_secrets_provider.py

@@ -0,0 +1,31 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.parameters import SecretsProvider
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+logger = Logger()
+# Create provider instance for more control
+secrets_provider = SecretsProvider()
+
+
+def lambda_handler(event, context: LambdaContext):
+    # Retrieve secrets with custom settings
+    secrets = secrets_provider.get_multiple(
+        names=["service/auth-token", "service/encryption-key"],
+        max_age=600,  # Cache for 10 minutes
+        transform="json",  # Parse JSON secrets
+        raise_on_transform_error=False,  # Don't fail on transform errors
+    )
+
+    # Handle potential transform failures
+    auth_token = secrets.get("service/auth-token")
+    encryption_key = secrets.get("service/encryption-key")
+
+    if auth_token is None:
+        logger.info("Warning: auth-token failed to parse as JSON")
+    if encryption_key is None:
+        logger.info("Warning: encryption-key failed to parse as JSON")
+
+    return {
+        "statusCode": 200,
+        "body": f"Retrieved {len([s for s in secrets.values() if s is not None])} valid secrets",
+    }