feat(parameters): add get_parameters_by_name for SSM params in distinct paths

aws_lambda_powertools/shared/functions.py

@@ -1,9 +1,10 @@
 import base64
+import itertools
 import logging
 import os
 import warnings
 from binascii import Error as BinAsciiError
-from typing import Optional, Union, overload
+from typing import Dict, Generator, Optional, Union, overload
 
 from aws_lambda_powertools.shared import constants
 
@@ -115,3 +116,8 @@ def powertools_debug_is_set() -> bool:
         return True
 
     return False
+
+
+def slice_dictionary(data: Dict, chunk_size: int) -> Generator[Dict, None, None]:
+    for _ in range(0, len(data), chunk_size):
+        yield {dict_key: data[dict_key] for dict_key in itertools.islice(data, chunk_size)}

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -15,8 +15,6 @@
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError, StoreClientError
 
-TRANSFORM_TYPE = "json"
-
 
 class AppConfigStore(StoreProvider):
     def __init__(
@@ -74,7 +72,7 @@ def get_raw_configuration(self) -> Dict[str, Any]:
                 dict,
                 self._conf_store.get(
                     name=self.name,
-                    transform=TRANSFORM_TYPE,
+                    transform="json",
                     max_age=self.cache_seconds,
                 ),
             )

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -9,7 +9,7 @@
 from .dynamodb import DynamoDBProvider
 from .exceptions import GetParameterError, TransformParameterError
 from .secrets import SecretsProvider, get_secret
-from .ssm import SSMProvider, get_parameter, get_parameters
+from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name
 
 __all__ = [
     "AppConfigProvider",
@@ -22,6 +22,7 @@
     "get_app_config",
     "get_parameter",
     "get_parameters",
+    "get_parameters_by_name",
     "get_secret",
     "clear_caches",
 ]

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -9,6 +9,8 @@
 import boto3
 from botocore.config import Config
 
+from aws_lambda_powertools.utilities.parameters.types import TransformOptions
+
 if TYPE_CHECKING:
     from mypy_boto3_appconfigdata import AppConfigDataClient
 
@@ -132,7 +134,7 @@ def get_app_config(
     name: str,
     environment: str,
     application: Optional[str] = None,
-    transform: Optional[str] = None,
+    transform: TransformOptions = None,
     force_fetch: bool = False,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     **sdk_options

aws_lambda_powertools/utilities/parameters/base.py

@@ -1,17 +1,31 @@
 """
 Base for Parameter providers
 """
+from __future__ import annotations
 
 import base64
 import json
 from abc import ABC, abstractmethod
-from collections import namedtuple
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Type, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    NamedTuple,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+    cast,
+    overload,
+)
 
 import boto3
 from botocore.config import Config
 
+from aws_lambda_powertools.utilities.parameters.types import TransformOptions
+
 from .exceptions import GetParameterError, TransformParameterError
 
 if TYPE_CHECKING:
@@ -22,37 +36,49 @@
 
 
 DEFAULT_MAX_AGE_SECS = 5
-ExpirableValue = namedtuple("ExpirableValue", ["value", "ttl"])
 # These providers will be dynamically initialized on first use of the helper functions
 DEFAULT_PROVIDERS: Dict[str, Any] = {}
 TRANSFORM_METHOD_JSON = "json"
 TRANSFORM_METHOD_BINARY = "binary"
 SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY]
 ParameterClients = Union["AppConfigDataClient", "SecretsManagerClient", "SSMClient"]
 
+TRANSFORM_METHOD_MAPPING = {
+    TRANSFORM_METHOD_JSON: json.loads,
+    TRANSFORM_METHOD_BINARY: base64.b64decode,
+    ".json": json.loads,
+    ".binary": base64.b64decode,
+    None: lambda x: x,
+}
+
+
+class ExpirableValue(NamedTuple):
+    value: str | bytes | Dict[str, Any]
+    ttl: datetime
+
 
 class BaseProvider(ABC):
     """
     Abstract Base Class for Parameter providers
     """
 
-    store: Any = None
+    store: Dict[Tuple[str, TransformOptions], ExpirableValue]
 
     def __init__(self):
         """
         Initialize the base provider
         """
 
-        self.store = {}
+        self.store: Dict[Tuple[str, TransformOptions], ExpirableValue] = {}
 
-    def _has_not_expired(self, key: Tuple[str, Optional[str]]) -> bool:
+    def has_not_expired_in_cache(self, key: Tuple[str, TransformOptions]) -> bool:
         return key in self.store and self.store[key].ttl >= datetime.now()
 
     def get(
         self,
         name: str,
         max_age: int = DEFAULT_MAX_AGE_SECS,
-        transform: Optional[str] = None,
+        transform: TransformOptions = None,
         force_fetch: bool = False,
         **sdk_options,
     ) -> Optional[Union[str, dict, bytes]]:
@@ -95,7 +121,7 @@ def get(
         value: Optional[Union[str, bytes, dict]] = None
         key = (name, transform)
 
-        if not force_fetch and self._has_not_expired(key):
+        if not force_fetch and self.has_not_expired_in_cache(key):
             return self.store[key].value
 
         try:
@@ -105,11 +131,11 @@ def get(
             raise GetParameterError(str(exc))
 
         if transform:
-            if isinstance(value, bytes):
-                value = value.decode("utf-8")
-            value = transform_value(value, transform)
+            value = transform_value(key=name, value=value, transform=transform, raise_on_transform_error=True)
 
-        self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))
+        # NOTE: don't cache None, as they might've been failed transforms and may be corrected
+        if value is not None:
+            self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))
 
         return value
 
@@ -124,7 +150,7 @@ def get_multiple(
         self,
         path: str,
         max_age: int = DEFAULT_MAX_AGE_SECS,
-        transform: Optional[str] = None,
+        transform: TransformOptions = None,
         raise_on_transform_error: bool = False,
         force_fetch: bool = False,
         **sdk_options,
@@ -160,8 +186,8 @@ def get_multiple(
         """
         key = (path, transform)
 
-        if not force_fetch and self._has_not_expired(key):
-            return self.store[key].value
+        if not force_fetch and self.has_not_expired_in_cache(key):
+            return self.store[key].value  # type: ignore # need to revisit entire typing here
 
         try:
             values = self._get_multiple(path, **sdk_options)
@@ -170,13 +196,8 @@ def get_multiple(
             raise GetParameterError(str(exc))
 
         if transform:
-            transformed_values: dict = {}
-            for (item, value) in values.items():
-                _transform = get_transform_method(item, transform)
-                if not _transform:
-                    continue
-                transformed_values[item] = transform_value(value, _transform, raise_on_transform_error)
-            values.update(transformed_values)
+            values.update(transform_value(values, transform, raise_on_transform_error))
+
         self.store[key] = ExpirableValue(values, datetime.now() + timedelta(seconds=max_age))
 
         return values
@@ -191,6 +212,12 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
     def clear_cache(self):
         self.store.clear()
 
+    def add_to_cache(self, key: Tuple[str, TransformOptions], value: Any, max_age: int):
+        if max_age <= 0:
+            return
+
+        self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))
+
     @staticmethod
     def _build_boto3_client(
         service_name: str,
@@ -258,57 +285,81 @@ def _build_boto3_resource_client(
         return session.resource(service_name=service_name, config=config, endpoint_url=endpoint_url)
 
 
-def get_transform_method(key: str, transform: Optional[str] = None) -> Optional[str]:
+def get_transform_method(value: str, transform: TransformOptions = None) -> Callable[..., Any]:
     """
     Determine the transform method
 
     Examples
     -------
-        >>> get_transform_method("key", "any_other_value")
+        >>> get_transform_method("key","any_other_value")
         'any_other_value'
-        >>> get_transform_method("key.json", "auto")
+        >>> get_transform_method("key.json","auto")
         'json'
-        >>> get_transform_method("key.binary", "auto")
+        >>> get_transform_method("key.binary","auto")
         'binary'
-        >>> get_transform_method("key", "auto")
+        >>> get_transform_method("key","auto")
         None
-        >>> get_transform_method("key", None)
+        >>> get_transform_method("key",None)
         None
 
     Parameters
     ---------
-    key: str
-        Only used when the tranform is "auto".
+    value: str
+        Only used when the transform is "auto".
     transform: str, optional
         Original transform method, only "auto" will try to detect the transform method by the key
 
     Returns
     ------
-    Optional[str]:
-        The transform method either when transform is "auto" then None, "json" or "binary" is returned
-        or the original transform method
+    Callable:
+        Transform function could be json.loads, base64.b64decode, or a lambda that echo the str value
     """
-    if transform != "auto":
-        return transform
+    transform_method = TRANSFORM_METHOD_MAPPING.get(transform)
+
+    if transform == "auto":
+        key_suffix = value.rsplit(".")[-1]
+        transform_method = TRANSFORM_METHOD_MAPPING.get(key_suffix, TRANSFORM_METHOD_MAPPING[None])
+
+    return cast(Callable, transform_method)  # https://github.com/python/mypy/issues/10740
+
+
+@overload
+def transform_value(
+    value: Dict[str, Any],
+    transform: TransformOptions,
+    raise_on_transform_error: bool = False,
+    key: str = "",
+) -> Dict[str, Any]:
+    ...
+
 
-    for transform_method in SUPPORTED_TRANSFORM_METHODS:
-        if key.endswith("." + transform_method):
-            return transform_method
-    return None
+@overload
+def transform_value(
+    value: Union[str, bytes, Dict[str, Any]],
+    transform: TransformOptions,
+    raise_on_transform_error: bool = False,
+    key: str = "",
+) -> Optional[Union[str, bytes, Dict[str, Any]]]:
+    ...
 
 
 def transform_value(
-    value: str, transform: str, raise_on_transform_error: Optional[bool] = True
-) -> Optional[Union[dict, bytes]]:
+    value: Union[str, bytes, Dict[str, Any]],
+    transform: TransformOptions,
+    raise_on_transform_error: bool = True,
+    key: str = "",
+) -> Optional[Union[str, bytes, Dict[str, Any]]]:
     """
-    Apply a transform to a value
+    Transform a value using one of the available options.
 
     Parameters
     ---------
     value: str
         Parameter value to transform
     transform: str
-        Type of transform, supported values are "json" and "binary"
+        Type of transform, supported values are "json", "binary", and "auto" based on suffix (.json, .binary)
+    key: str
+        Parameter key when transform is auto to infer its transform method
     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
@@ -318,18 +369,41 @@ def transform_value(
     TransformParameterError:
         When the parameter value could not be transformed
     """
+    # Maintenance: For v3, we should consider returning the original value for soft transform failures.
+
+    err_msg = "Unable to transform value using '{transform}' transform: {exc}"
+
+    if isinstance(value, bytes):
+        value = value.decode("utf-8")
+
+    if isinstance(value, dict):
+        # NOTE: We must handle partial failures when receiving multiple values
+        # where one of the keys might fail during transform, e.g. `{"a": "valid", "b": "{"}`
+        # expected: `{"a": "valid", "b": None}`
+
+        transformed_values: Dict[str, Any] = {}
+        for dict_key, dict_value in value.items():
+            transform_method = get_transform_method(value=dict_key, transform=transform)
+            try:
+                transformed_values[dict_key] = transform_method(dict_value)
+            except Exception as exc:
+                if raise_on_transform_error:
+                    raise TransformParameterError(err_msg.format(transform=transform, exc=exc)) from exc
+                transformed_values[dict_key] = None
+        return transformed_values
+
+    if transform == "auto":
+        # key="a.json", value='{"a": "b"}', or key="a.binary", value="b64_encoded"
+        transform_method = get_transform_method(value=key, transform=transform)
+    else:
+        # value='{"key": "value"}
+        transform_method = get_transform_method(value=value, transform=transform)
 
     try:
-        if transform == TRANSFORM_METHOD_JSON:
-            return json.loads(value)
-        elif transform == TRANSFORM_METHOD_BINARY:
-            return base64.b64decode(value)
-        else:
-            raise ValueError(f"Invalid transform type '{transform}'")
-
+        return transform_method(value)
     except Exception as exc:
         if raise_on_transform_error:
-            raise TransformParameterError(str(exc))
+            raise TransformParameterError(err_msg.format(transform=transform, exc=exc)) from exc
         return None