feat: add parameter utility

aws_lambda_powertools/utilities/__init__.py

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+
+"""General utilities for Powertools"""

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+
+"""
+Parameter retrieval and caching utility
+"""
+
+from .base import BaseProvider
+from .dynamodb import DynamoDBProvider
+from .exceptions import GetParameterError
+from .secrets import SecretsProvider, get_secret
+from .ssm import SSMProvider, get_parameter, get_parameters
+
+__all__ = [
+    "BaseProvider",
+    "GetParameterError",
+    "DynamoDBProvider",
+    "SecretsProvider",
+    "SSMProvider",
+    "get_parameter",
+    "get_parameters",
+    "get_secret",
+]

aws_lambda_powertools/utilities/parameters/base.py

@@ -0,0 +1,124 @@
+"""
+Base for Parameter providers
+"""
+
+import base64
+import json
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime, timedelta
+from typing import Dict, Optional, Union
+
+from .exceptions import GetParameterError
+
+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 = {}
+
+
+class BaseProvider(ABC):
+    """
+    Abstract Base Class for Parameter providers
+    """
+
+    store = None
+
+    def __init__(self):
+        """
+        Initialize the base provider
+        """
+
+        self.store = {}
+
+    def get(
+        self, name: str, max_age: int = DEFAULT_MAX_AGE_SECS, transform: Optional[str] = None, **kwargs
+    ) -> Union[str, list, dict, bytes]:
+        """
+        Retrieve a parameter value or return the cached value
+
+        Parameters
+        ----------
+
+        name: str
+            Parameter name
+        max_age: int
+            Maximum age of the cached value
+        transform: str
+            Optional transformation of the parameter value. Supported values
+            are "json" for JSON strings and "binary" for base 64 encoded
+            values.
+
+        Raises
+        ------
+
+        GetParameterError
+            When the parameter provider fails to retrieve a parameter value for
+            a given name.
+        """
+
+        # If there are multiple calls to the same parameter but in a different
+        # transform, they will be stored multiple times. This allows us to
+        # optimize by transforming the data only once per retrieval, thus there
+        # is no need to transform cached values multiple times. However, this
+        # means that we need to make multiple calls to the underlying parameter
+        # store if we need to return it in different transforms. Since the number
+        # of supported transform is small and the probability that a given
+        # parameter will always be used in a specific transform, this should be
+        # an acceptable tradeoff.
+        key = (name, transform)
+
+        if key not in self.store or self.store[key].ttl < datetime.now():
+            try:
+                value = self._get(name, **kwargs)
+            # Encapsulate all errors into a generic GetParameterError
+            except Exception as exc:
+                raise GetParameterError(str(exc))
+
+            if transform == "json":
+                value = json.loads(value)
+            elif transform == "binary":
+                value = base64.b64decode(value)
+
+            self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age),)
+
+        return self.store[key].value
+
+    @abstractmethod
+    def _get(self, name: str, **kwargs) -> str:
+        """
+        Retrieve paramater value from the underlying parameter store
+        """
+        raise NotImplementedError()
+
+    def get_multiple(
+        self, path: str, max_age: int = DEFAULT_MAX_AGE_SECS, transform: Optional[str] = None, **kwargs
+    ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
+        """
+        Retrieve multiple parameters based on a path prefix
+        """
+
+        key = (path, transform)
+
+        if key not in self.store or self.store[key].ttl < datetime.now():
+            try:
+                values = self._get_multiple(path, **kwargs)
+            # Encapsulate all errors into a generic GetParameterError
+            except Exception as exc:
+                raise GetParameterError(str(exc))
+
+            if transform == "json":
+                values = {k: json.loads(v) for k, v in values.items()}
+            elif transform == "binary":
+                values = {k: base64.b64decode(v) for k, v in values.items()}
+
+            self.store[key] = ExpirableValue(values, datetime.now() + timedelta(seconds=max_age),)
+
+        return self.store[key].value
+
+    @abstractmethod
+    def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+        """
+        Retrieve multiple parameter values from the underlying parameter store
+        """
+        raise NotImplementedError()

aws_lambda_powertools/utilities/parameters/dynamodb.py

@@ -0,0 +1,164 @@
+"""
+Amazon DynamoDB parameter retrieval and caching utility
+"""
+
+
+from typing import Dict, Optional
+
+import boto3
+from boto3.dynamodb.conditions import Key
+from botocore.config import Config
+
+from .base import BaseProvider
+
+
+class DynamoDBProvider(BaseProvider):
+    """
+    Amazon DynamoDB Parameter Provider
+
+    Parameters
+    ----------
+    table_name: str
+        Name of the DynamoDB table that stores parameters
+    key_attr: str, optional
+        Hash key for the DynamoDB table
+    value_attr: str, optional
+        Attribute that contains the values in the DynamoDB table
+    config: botocore.config.Config, optional
+        Botocore configuration to pass during client initialization
+
+    Example
+    -------
+    **Retrieves a parameter value from a DynamoDB table**
+
+    In this example, the DynamoDB table uses `id` as hash key and stores the value in the `value`
+    attribute.
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider("ParametersTable")
+        >>>
+        >>> ddb_provider.get("my-parameter")
+
+    **Retrieves a parameter value from a DynamoDB table that has custom attribute names**
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider(
+        ...     "ParametersTable",
+        ...     key_attr="my-id",
+        ...     value_attr="my-value"
+        ... )
+        >>>
+        >>> ddb_provider.get("my-parameter")
+
+    **Retrieves a parameter value from a DynamoDB table in another AWS region**
+
+        >>> from botocore.config import Config
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>>
+        >>> config = Config(region_name="us-west-1")
+        >>> ddb_provider = DynamoDBProvider("ParametersTable", config=config)
+        >>>
+        >>> 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
+    the hood. This expects that the sort key is named `sk`.
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider("ParametersTable")
+        >>>
+        >>> ddb_provider.get_multiple("my-parameters")
+
+    **Retrieves multiple values from a DynamoDB table with a custom sort key**
+
+    In this case, the provider will use a sort key to retrieve multiple values using a query under
+    the hood.
+
+        >>> from aws_lambda_powertools.utilities.parameters import DynamoDBProvider
+        >>> ddb_provider = DynamoDBProvider("ParametersTable")
+        >>>
+        >>> 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
+    key_attr = None
+    value_attr = None
+
+    def __init__(
+        self, table_name: str, key_attr: str = "id", value_attr: str = "value", config: Optional[Config] = None,
+    ):
+        """
+        Initialize the DynamoDB client
+        """
+
+        config = config or Config()
+        self.table = boto3.resource("dynamodb", config=config).Table(table_name)
+
+        self.key_attr = key_attr
+        self.value_attr = value_attr
+
+        super().__init__()
+
+    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, optional
+            Dictionary of options that will be passed to the get_item call
+        """
+
+        # 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", **sdk_options) -> Dict[str, str]:
+        """
+        Retrieve multiple parameter values from Amazon DynamoDB
+
+        Parameters
+        ----------
+        path: str
+            Path to retrieve the parameters
+        sort_attr: str, optional
+            Name of the DynamoDB table sort key (defaults to 'sk')
+        sdk_options: dict, optional
+            Dictionary of options that will be passed to the query call
+        """
+
+        # 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:
+            sdk_options["ExclusiveStartKey"] = response["LastEvaluatedKey"]
+            response = self.table.query(**sdk_options)
+            items.extend(response.get("Items", []))
+
+        retval = {}
+        for item in items:
+            retval[item[sort_attr]] = item[self.value_attr]
+
+        return retval

aws_lambda_powertools/utilities/parameters/exceptions.py

@@ -0,0 +1,7 @@
+"""
+Parameter retrieval exceptions
+"""
+
+
+class GetParameterError(Exception):
+    """When a provider raises an exception on parameter retrieval"""