feat: Add AppConfig parameter provider

aws_lambda_powertools/utilities/parameters/__init__.py

@@ -4,19 +4,22 @@
 Parameter retrieval and caching utility
 """
 
+from .appconfig 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",

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -0,0 +1,158 @@
+"""
+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
+    ----------
+    environment: str
+        Environment of the configuration to pass during client initialization
+    application: str, optional
+        Application of the configuration to pass during client initialization
+    config: botocore.config.Config, optional
+        Botocore configuration to pass during client initialization
+
+    Example
+    -------
+    **Retrieves the latest configuration value from App Config**
+
+        >>> from aws_lambda_powertools.utilities.parameters import AppConfigProvider
+        >>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app")
+        >>>
+        >>> value : bytes = appconf_provider.get("my_conf")
+        >>>
+        >>> print(value)
+        My configuration value
+
+    **Retrieves a configuration value from App Config in another AWS region**
+
+        >>> from botocore.config import Config
+        >>> from aws_lambda_powertools.utilities.parameters import AppConfigProvider
+        >>>
+        >>> config = Config(region_name="us-west-1")
+        >>> appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app", config=config)
+        >>>
+        >>> value : bytes = appconf_provider.get("my_conf")
+        >>>
+        >>> print(value)
+        My configuration value
+
+    """
+
+    client = None
+
+    def __init__(
+        self, environment: str, application: Optional[str] = None, config: Optional[Config] = None,
+    ):
+        """
+        Initialize the App Config 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
+        self.current_version = ""
+
+        super().__init__()
+
+    def _get(self, name: str, **sdk_options) -> str:
+        """
+        Retrieve a parameter value from AWS App config.
+
+        Parameters
+        ----------
+        name: str
+            Name of the configuration
+        environment: str
+            Environment of the configuration
+        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
+
+        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 App Config Provider
+        """
+        raise NotImplementedError()
+
+
+def get_app_config(
+    name: str, environment: str, application: Optional[str] = None, transform: Optional[str] = None, **sdk_options
+) -> Union[str, list, dict, bytes]:
+    """
+    Retrieve a configuration value from AWS App Config.
+
+    Parameters
+    ----------
+    name: str
+        Name of the configuration
+    environment: str
+        Environment of the configuration
+    application: str
+        Application of the configuration
+    transform: str, optional
+        Transforms the content from a JSON object ('json') or base64 binary string ('binary')
+    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 the latest version of configuration value from App Config**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_app_config
+        >>>
+        >>> value = get_app_config("my_config", environment="my_env", application="my_env")
+        >>>
+        >>> print(value)
+        My configuration value
+
+    **Retrieves a confiugration value and decodes it using a JSON decoder**
+
+        >>> from aws_lambda_powertools.utilities.parameters import get_parameter
+        >>>
+        >>> value = get_app_config("my_config", environment="my_env", application="my_env", transform='json')
+        >>>
+        >>> print(value)
+        My configuration's JSON 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
+
+    return DEFAULT_PROVIDERS["appconfig"].get(name, transform=transform, **sdk_options)

aws_lambda_powertools/utilities/parameters/base.py

@@ -17,6 +17,7 @@
 DEFAULT_PROVIDERS = {}
 TRANSFORM_METHOD_JSON = "json"
 TRANSFORM_METHOD_BINARY = "binary"
+TRANSFORM_METHOD_YAML = "yaml"
 SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY]
 
 
@@ -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:
+            raise NotImplementedError
         else:
             raise ValueError(f"Invalid transform type '{transform}'")
 

docs/content/utilities/parameters.mdx

@@ -24,6 +24,7 @@ SSM Parameter Store | `get_parameters`, `SSMProvider.get_multiple` | `ssm:GetPar
 Secrets Manager | `get_secret`, `SecretsManager.get` | `secretsmanager:GetSecretValue`
 DynamoDB | `DynamoDBProvider.get` | `dynamodb:GetItem`
 DynamoDB | `DynamoDBProvider.get_multiple` | `dynamodb:Query`
+App Config | `AppConfigProvider.get_app_config`, `get_app_config` | `appconfig:GetConfiguration`
 
 ## SSM Parameter Store
 
@@ -204,6 +205,37 @@ def handler(event, context):
     value = dynamodb_provider.get("my-parameter")
 ```
 
+## App Config
+
+For configurations stored in App Config, use `get_app_config`.
+The following will retrieve the latest version and store it in the cache.
+
+```python:title=appconfig.py
+from aws_lambda_powertools.utilities import parameters
+
+def handler(event, context):
+    # Retrieve a single configuration, latest version
+    value: bytes = parameters.get_app_config(name="my_configuration", environment="my_env", application="my_app")
+```
+
+### AppConfigProvider class
+
+Alternatively, you can use the `AppConfigProvider` class, which give more flexibility, such as the ability to configure the underlying SDK client.
+
+This can be used to retrieve values from other regions, change the retry behavior, etc.
+
+```python:title=appconfig.py
+from aws_lambda_powertools.utilities import parameters
+from botocore.config import Config
+
+config = Config(region_name="us-west-1")
+appconf_provider = parameters.AppConfigProvider(environment="my_env", application="my_app", config=config)
+
+def handler(event, context):
+    # Retrieve a single secret
+    value : bytes = appconf_provider.get("my_conf")
+```
+
 ## Create your own provider
 
 You can create your own custom parameter store provider by inheriting the `BaseProvider` class, and implementing both `_get()` and `_get_multiple()` methods to retrieve a single, or multiple parameters from your custom store.
@@ -334,13 +366,14 @@ def handler(event, context):
 
 Here is the mapping between this utility's functions and methods and the underlying SDK:
 
-| Provider            | Function/Method                 | Client name | Function name |
-|---------------------|---------------------------------|-------------|---------------|
-| SSM Parameter Store | `get_parameter`                 | `ssm`       | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter) |
-| SSM Parameter Store | `get_parameters`                | `ssm`       | [get_parameters_by_path](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameters_by_path) |
-| SSM Parameter Store | `SSMProvider.get`               | `ssm`       | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter) |
-| 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) |
+| Provider            | Function/Method                 | Client name      | Function name |
+|---------------------|---------------------------------|------------------|----------------|
+| SSM Parameter Store | `get_parameter`                 | `ssm`            | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter) |
+| SSM Parameter Store | `get_parameters`                | `ssm`            | [get_parameters_by_path](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameters_by_path) |
+| SSM Parameter Store | `SSMProvider.get`               | `ssm`            | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter) |
+| 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) |
 | 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) |
 | Secrets Manager     | `SecretsManager.get`            | `secretsmanager` | [get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.get_secret_value) |
-| DynamoDB            | `DynamoDBProvider.get`          | `dynamodb` ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [get_item](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.get_item)
-| DynamoDB            | `DynamoDBProvider.get_multiple` | `dynamodb` ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [query](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.query)
+| DynamoDB            | `DynamoDBProvider.get`          | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [get_item](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.get_item)
+| DynamoDB            | `DynamoDBProvider.get_multiple` | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [query](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.query)
+| App Config           | `get_app_config`                 | `appconfig`       | [get_configuration](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfig.html#AppConfig.Client.get_configuration) |

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,87 @@ 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, config):
+    """
+    Test get_configuration.get with default values
+    """
+
+    # Create a new provider
+    environment = "dev"
+    application = "myapp"
+    provider = parameters.AppConfigProvider(environment=environment, application=application, config=config)
+
+    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, transform="json", ClientConfigurationVersion="2")
+
+        assert value == mock_body_json
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_appconf_provider_get_configuration_no_transform(mock_name, config):
+    """
+    Test appconfigprovider.get with default values
+    """
+
+    # Create a new provider
+    environment = "dev"
+    application = "myapp"
+    provider = parameters.AppConfigProvider(environment=environment, application=application, config=config)
+
+    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)
+        str_value = value.decode("utf-8")
+        assert str_value == json.dumps(mock_body_json)
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_appconf_get_app_config_no_transform(monkeypatch, mock_name):
+    """
+    Test get_app_config()
+    """
+    mock_body_json = {"myenvvar1": "Black Panther", "myenvvar2": 3}
+
+    class TestProvider(BaseProvider):
+        def _get(self, name: str, **kwargs) -> str:
+            assert name == mock_name
+            return json.dumps(mock_body_json).encode("utf-8")
+
+        def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+            raise NotImplementedError()
+
+    monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "appconfig", TestProvider())
+
+    environment = "dev"
+    application = "myapp"
+    value = parameters.get_app_config(mock_name, environment=environment, application=application)
+    str_value = value.decode("utf-8")
+    assert str_value == json.dumps(mock_body_json)
+
+
 def test_transform_value_json(mock_value):
     """
     Test transform_value() with a json transform