feat: Add AppConfig parameter provider

Author: Ran Isenberg

Diff for: aws_lambda_powertools/utilities/parameters/__init__.py

@@ -4,19 +4,22 @@
 Parameter retrieval and caching utility
 """
 
+from .appconf import AppConfigProvider, get_app_config
 from .base import BaseProvider
 from .dynamodb import DynamoDBProvider
 from .exceptions import GetParameterError, TransformParameterError
 from .secrets import SecretsProvider, get_secret
 from .ssm import SSMProvider, get_parameter, get_parameters
 
 __all__ = [
+    "AppConfigProvider",
     "BaseProvider",
     "GetParameterError",
     "DynamoDBProvider",
     "SecretsProvider",
     "SSMProvider",
     "TransformParameterError",
+    "get_app_config",
     "get_parameter",
     "get_parameters",
     "get_secret",

Diff for: aws_lambda_powertools/utilities/parameters/appconf.py

@@ -0,0 +1,185 @@
+"""
+AWS App Config configuration retrieval and caching utility
+"""
+
+
+import os
+from typing import Dict, Optional, Union
+from uuid import uuid4
+
+import boto3
+from botocore.config import Config
+
+from .base import DEFAULT_PROVIDERS, BaseProvider
+
+CLIENT_ID = str(uuid4())
+
+
+class AppConfigProvider(BaseProvider):
+    """
+    AWS App Config Provider
+
+    Parameters
+    ----------
+    config: botocore.config.Config, optional
+        Botocore configuration to pass during client initialization
+
+    Example
+    -------
+    **Retrieves a parameter value from App Config**
+
+        >>> from aws_lambda_powertools.utilities.parameters import AppConfigProvider
+        >>> appconfig_provider = AppConfigProvider()
+        >>>
+        >>> value: Dict[str, str] = appconfig_provider.get("my-configuration")
+        >>>
+        >>> print(value)
+        My configuration values
+
+    **Retrieves a parameter value from Systems Manager Parameter Store in another AWS region**
+
+        >>> from botocore.config import Config
+        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
+        >>>
+        >>> config = Config(region_name="us-west-1")
+        >>> ssm_provider = SSMProvider(config=config)
+        >>>
+        >>> value = ssm_provider.get("/my/parameter")
+        >>>
+        >>> print(value)
+        My parameter value
+
+    **Retrieves multiple parameter values from Systems Manager Parameter Store using a path prefix**
+
+        >>> from aws_lambda_powertools.utilities.parameters import SSMProvider
+        >>> ssm_provider = SSMProvider()
+        >>>
+        >>> values = ssm_provider.get_multiple("/my/path/prefix")
+        >>>
+        >>> for key, value in values.items():
+        ...     print(key, value)
+        /my/path/prefix/a   Parameter value a
+        /my/path/prefix/b   Parameter value b
+        /my/path/prefix/c   Parameter value c
+
+    **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()
+        >>>
+        >>> values = ssm_provider.get_multiple("/my/path/prefix", MaxResults=10)
+        >>>
+        >>> for key, value in values.items():
+        ...     print(key, value)
+        /my/path/prefix/a   Parameter value a
+        /my/path/prefix/b   Parameter value b
+        /my/path/prefix/c   Parameter value c
+    """
+
+    client = None
+
+    def __init__(
+        self, environment: str, application: Optional[str] = None, config: Optional[Config] = None,
+    ):
+        """
+        Initialize the SSM Parameter Store client
+        """
+
+        config = config or Config()
+        self.client = boto3.client("appconfig", config=config)
+        self.application = application or os.getenv("POWERTOOLS_SERVICE_NAME") or "application_undefined"
+        self.environment = environment
+
+        super().__init__()
+
+    def _get(self, name: str, client_conf_version: Optional[str] = None, **sdk_options) -> str:
+        """
+        Retrieve a parameter value from AWS Systems Manager Parameter Store
+
+        Parameters
+        ----------
+        name: str
+            Parameter name
+        decrypt: bool, optional
+            If the parameter value should be decrypted
+        sdk_options: dict, optional
+            Dictionary of options that will be passed to the Parameter Store get_parameter API call
+        """
+
+        sdk_options["Configuration"] = name
+        sdk_options["Application"] = self.application
+        sdk_options["Environment"] = self.environment
+        sdk_options["ClientId"] = CLIENT_ID
+        if client_conf_version is not None:
+            sdk_options["ClientConfigurationVersion"] = client_conf_version
+
+        response = self.client.get_configuration(**sdk_options)
+        return response["Content"].read()  # read() of botocore.response.StreamingBody
+
+    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
+        """
+        Retrieving multiple parameter values is not supported with AWS Secrets Manager
+        """
+        raise NotImplementedError()
+
+
+def get_app_config(
+    name: str,
+    environment: str,
+    application: Optional[str] = None,
+    transform: Optional[str] = None,
+    client_conf_version: Optional[str] = None,
+    **sdk_options
+) -> Union[str, list, dict, bytes]:
+    """
+    Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store
+
+    Parameters
+    ----------
+    name: str
+        Name of the parameter
+    transform: str, optional
+        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
+    decrypt: bool, optional
+        If the parameter values should be decrypted
+    sdk_options: dict, optional
+        Dictionary of options that will be passed to the Parameter Store get_parameter API call
+
+    Raises
+    ------
+    GetParameterError
+        When the parameter provider fails to retrieve a parameter value for
+        a given name.
+    TransformParameterError
+        When the parameter provider fails to transform a parameter value.
+
+    Example
+    -------
+    **Retrieves a parameter value from Systems Manager Parameter Store**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
+        >>>
+        >>> value = get_parameter("/my/parameter")
+        >>>
+        >>> print(value)
+        My parameter value
+
+    **Retrieves a parameter value and decodes it using a Base64 decoder**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
+        >>>
+        >>> value = get_parameter("/my/parameter", transform='binary')
+        >>>
+        >>> print(value)
+        My parameter value
+    """
+
+    # Only create the provider if this function is called at least once
+    if "appconfig" not in DEFAULT_PROVIDERS:
+        DEFAULT_PROVIDERS["appconfig"] = AppConfigProvider(environment=environment, application=application)
+
+        sdk_options["ClientId"] = CLIENT_ID
+        if client_conf_version is not None:
+            sdk_options["ClientConfigurationVersion"] = client_conf_version
+
+    return DEFAULT_PROVIDERS["appconfig"].get(name, transform=transform, **sdk_options)

Diff for: aws_lambda_powertools/utilities/parameters/base.py

@@ -17,7 +17,8 @@
 DEFAULT_PROVIDERS = {}
 TRANSFORM_METHOD_JSON = "json"
 TRANSFORM_METHOD_BINARY = "binary"
-SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY]
+TRANSFORM_METHOD_YAML = "yaml"
+SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY, TRANSFORM_METHOD_YAML]
 
 
 class BaseProvider(ABC):
@@ -230,6 +231,8 @@ def transform_value(value: str, transform: str, raise_on_transform_error: bool =
             return json.loads(value)
         elif transform == TRANSFORM_METHOD_BINARY:
             return base64.b64decode(value)
+        elif transform == TRANSFORM_METHOD_YAML:
+            return value  # todo
         else:
             raise ValueError(f"Invalid transform type '{transform}'")
 

Diff for: tests/functional/test_utilities_parameters.py

@@ -3,12 +3,14 @@
 import random
 import string
 from datetime import datetime, timedelta
+from io import BytesIO
 from typing import Dict
 
 import pytest
 from boto3.dynamodb.conditions import Key
 from botocore import stub
 from botocore.config import Config
+from botocore.response import StreamingBody
 
 from aws_lambda_powertools.utilities import parameters
 from aws_lambda_powertools.utilities.parameters.base import BaseProvider, ExpirableValue
@@ -1451,6 +1453,35 @@ def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
     assert value == mock_value
 
 
+def test_appconf_provider_get_configuration_json_content_type(mock_name, mock_value, config):
+    """
+    Test get_configuration.get with default values
+    """
+
+    # Create a new provider
+    environment = "dev"
+    application = "myapp"
+    provider = parameters.AppConfigProvider(environment=environment, application=application)
+
+    mock_body_json = {"myenvvar1": "Black Panther", "myenvvar2": 3}
+    encoded_message = json.dumps(mock_body_json).encode("utf-8")
+    mock_value = StreamingBody(BytesIO(encoded_message), len(encoded_message))
+
+    # Stub the boto3 client
+    stubber = stub.Stubber(provider.client)
+    response = {"Content": mock_value, "ConfigurationVersion": "1", "ContentType": "application/json"}
+    stubber.add_response("get_configuration", response)
+    stubber.activate()
+
+    try:
+        value = provider.get(mock_name, client_conf_version="1", transform="json")
+
+        assert value == mock_body_json
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
 def test_transform_value_json(mock_value):
     """
     Test transform_value() with a json transform