feat: add override SDK options for parameter utilities

Author: nmoutschen

aws_lambda_powertools/utilities/parameters/dynamodb.py

@@ -49,6 +49,13 @@ class DynamoDBProvider(BaseProvider):
         >>>
         >>> ddb_provider.get("my-parameter")
 
+    **Retrieves a parameter value from a DynamoDB table passing options to the SDK call**
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider("ParametersTable")
+        >>>
+        >>> ddb_provider.get("my-parameter", ConsistentRead=True)
+
     **Retrieves multiple values from a DynamoDB table**
 
     In this case, the provider will use a sort key to retrieve multiple values using a query under
@@ -69,6 +76,12 @@ class DynamoDBProvider(BaseProvider):
         >>>
         >>> ddb_provider.get_multiple("my-parameters", sort_attr="my-sort-attr")
 
+    **Retrieves multiple values from a DynamoDB table passing options to the SDK calls**
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider("ParametersTable")
+        >>>
+        >>> ddb_provider.get("my-parameter", ConsistentRead=True)
     """
 
     table = None
@@ -90,14 +103,24 @@ def __init__(
 
         super().__init__()
 
-    def _get(self, name: str, **kwargs) -> str:
+    def _get(self, name: str, **sdk_options) -> str:
         """
         Retrieve a parameter value from Amazon DynamoDB
+
+        Parameters
+        ----------
+        name: str
+            Name of the parameter
+        sdk_options: dict
+            Dictionary of options that will be passed to the get_item call
         """
 
-        return self.table.get_item(Key={self.key_attr: name})["Item"][self.value_attr]
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["Key"] = {self.key_attr: name}
+
+        return self.table.get_item(**sdk_options)["Item"][self.value_attr]
 
-    def _get_multiple(self, path: str, sort_attr: str = "sk", **kwargs) -> Dict[str, str]:
+    def _get_multiple(self, path: str, sort_attr: str = "sk", **sdk_options) -> Dict[str, str]:
         """
         Retrieve multiple parameter values from Amazon DynamoDB
 
@@ -107,16 +130,20 @@ def _get_multiple(self, path: str, sort_attr: str = "sk", **kwargs) -> Dict[str,
             Path to retrieve the parameters
         sort_attr: str
             Name of the DynamoDB table sort key (defaults to 'sk')
+        sdk_options: dict
+            Dictionary of options that will be passed to the query call
         """
 
-        response = self.table.query(KeyConditionExpression=Key(self.key_attr).eq(path))
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["KeyConditionExpression"] = Key(self.key_attr).eq(path)
+
+        response = self.table.query(**sdk_options)
         items = response.get("Items", [])
 
         # Keep querying while there are more items matching the partition key
         while "LastEvaluatedKey" in response:
-            response = self.table.query(
-                KeyConditionExpression=Key(self.key_attr).eq(path), ExclusiveStartKey=response["LastEvaluatedKey"],
-            )
+            sdk_options["ExclusiveStartKey"] = response["LastEvaluatedKey"]
+            response = self.table.query(**sdk_options)
             items.extend(response.get("Items", []))
 
         retval = {}

aws_lambda_powertools/utilities/parameters/secrets.py

@@ -33,6 +33,13 @@ class SecretsProvider(BaseProvider):
         >>> secrets_provider = SecretsProvider(config=config)
         >>>
         >>> secrets_provider.get("my-parameter")
+
+    **Retrieves a parameter value from Secrets Manager passing options to the SDK call**
+
+        >>> from aws_lambda_powertools.utilities.parameters import SecretsProvider
+        >>> secrets_provider = SecretsProvider()
+        >>>
+        >>> secrets_provider.get("my-parameter", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
     """
 
     client = None
@@ -48,14 +55,24 @@ def __init__(self, config: Optional[Config] = None):
 
         super().__init__()
 
-    def _get(self, name: str, **kwargs) -> str:
+    def _get(self, name: str, **sdk_options) -> str:
         """
         Retrieve a parameter value from AWS Systems Manager Parameter Store
+
+        Parameters
+        ----------
+        name: str
+            Name of the parameter
+        sdk_options: dict
+            Dictionary of options that will be passed to the get_secret_value call
         """
 
-        return self.client.get_secret_value(SecretId=name)["SecretString"]
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["SecretId"] = name
+
+        return self.client.get_secret_value(**sdk_options)["SecretString"]
 
-    def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
         """
         Retrieving multiple parameter values is not supported with AWS Secrets Manager
         """

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -34,12 +34,19 @@ class SSMProvider(BaseProvider):
         >>>
         >>> ssm_provider.get("/my/parameter")
 
-    **Retrieves multiple parameter values from Systes Manager Parameter Store using a path prefix**
+    **Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix**
 
         >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
         >>> ssm_provider = SSMProvider()
         >>>
         >>> ssm_provider.get_multiple("/my/path/prefix")
+
+    **Retrieves multiple parameter values from Systems Manager Parameter Store passing options to the SDK call**
+
+        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
+        >>> ssm_provider = SSMProvider()
+        >>>
+        >>> ssm_provider.get_multiple("/my/path/prefix", MaxResults=10)
     """
 
     client = None
@@ -56,7 +63,7 @@ def __init__(
 
         super().__init__()
 
-    def _get(self, name: str, decrypt: bool = False, **kwargs) -> str:
+    def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str:
         """
         Retrieve a parameter value from AWS Systems Manager Parameter Store
 
@@ -66,11 +73,17 @@ def _get(self, name: str, decrypt: bool = False, **kwargs) -> str:
             Parameter name
         decrypt: bool
             If the parameter value should be decrypted
+        sdk_options: dict
+            Dictionary of options that will be passed to the get_parameter call
         """
 
-        return self.client.get_parameter(Name=name, WithDecryption=decrypt)["Parameter"]["Value"]
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["Name"] = name
+        sdk_options["WithDecryption"] = decrypt
+
+        return self.client.get_parameter(**sdk_options)["Parameter"]["Value"]
 
-    def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = False, **kwargs) -> Dict[str, str]:
+    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
 
@@ -82,9 +95,16 @@ def _get_multiple(self, path: str, decrypt: bool = False, recursive: bool = Fals
             If the parameter values should be decrypted
         recursive: bool
             If this should retrieve the parameter values recursively or not
+        sdk_options: dict
+            Dictionary of options that will be passed to the get_parameters_by_path call
         """
 
-        response = self.client.get_parameters_by_path(Path=path, WithDecryption=decrypt, Recursive=recursive)
+        # Explicit arguments will take precedence over keyword arguments
+        sdk_options["Path"] = path
+        sdk_options["WithDecryption"] = decrypt
+        sdk_options["Recursive"] = recursive
+
+        response = self.client.get_parameters_by_path(**sdk_options)
         parameters = response.get("Parameters", [])
 
         # Keep retrieving parameters