refactor(feature-flags): optimize UX and maintenance

.pre-commit-config.yaml

@@ -24,7 +24,7 @@ repos:
             types: [python]
           - id: isort
             name: formatting::isort
-            entry: poetry run isort -rc
+            entry: poetry run isort
             language: system
             types: [python]
     - repo: local

aws_lambda_powertools/shared/jmespath_utils.py

@@ -0,0 +1,55 @@
+import base64
+import gzip
+import json
+from typing import Any, Dict, Optional, Union
+
+import jmespath
+from jmespath.exceptions import LexerError
+
+from aws_lambda_powertools.utilities.validation import InvalidEnvelopeExpressionError
+from aws_lambda_powertools.utilities.validation.base import logger
+
+
+class PowertoolsFunctions(jmespath.functions.Functions):
+    @jmespath.functions.signature({"types": ["string"]})
+    def _func_powertools_json(self, value):
+        return json.loads(value)
+
+    @jmespath.functions.signature({"types": ["string"]})
+    def _func_powertools_base64(self, value):
+        return base64.b64decode(value).decode()
+
+    @jmespath.functions.signature({"types": ["string"]})
+    def _func_powertools_base64_gzip(self, value):
+        encoded = base64.b64decode(value)
+        uncompressed = gzip.decompress(encoded)
+
+        return uncompressed.decode()
+
+
+def unwrap_event_from_envelope(data: Union[Dict, str], envelope: str, jmespath_options: Optional[Dict]) -> Any:
+    """Searches data using JMESPath expression
+
+    Parameters
+    ----------
+    data : Dict
+        Data set to be filtered
+    envelope : str
+        JMESPath expression to filter data against
+    jmespath_options : Dict
+        Alternative JMESPath options to be included when filtering expr
+
+    Returns
+    -------
+    Any
+        Data found using JMESPath expression given in envelope
+    """
+    if not jmespath_options:
+        jmespath_options = {"custom_functions": PowertoolsFunctions()}
+
+    try:
+        logger.debug(f"Envelope detected: {envelope}. JMESPath options: {jmespath_options}")
+        return jmespath.search(envelope, data, options=jmespath.Options(**jmespath_options))
+    except (LexerError, TypeError, UnicodeError) as e:
+        message = f"Failed to unwrap event from envelope using expression. Error: {e} Exp: {envelope}, Data: {data}"  # noqa: B306, E501
+        raise InvalidEnvelopeExpressionError(message)

aws_lambda_powertools/utilities/feature_flags/__init__.py

@@ -0,0 +1,15 @@
+"""Advanced feature flags utility"""
+from .appconfig import AppConfigStore
+from .base import StoreProvider
+from .exceptions import ConfigurationStoreError
+from .feature_flags import FeatureFlags
+from .schema import RuleAction, SchemaValidator
+
+__all__ = [
+    "ConfigurationStoreError",
+    "FeatureFlags",
+    "RuleAction",
+    "SchemaValidator",
+    "AppConfigStore",
+    "StoreProvider",
+]

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -0,0 +1,92 @@
+import logging
+import traceback
+from typing import Any, Dict, Optional, cast
+
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
+
+from ...shared import jmespath_utils
+from .base import StoreProvider
+from .exceptions import ConfigurationStoreError, StoreClientError
+
+logger = logging.getLogger(__name__)
+
+TRANSFORM_TYPE = "json"
+
+
+class AppConfigStore(StoreProvider):
+    def __init__(
+        self,
+        environment: str,
+        application: str,
+        name: str,
+        cache_seconds: int,
+        sdk_config: Optional[Config] = None,
+        envelope: str = "",
+        jmespath_options: Optional[Dict] = None,
+    ):
+        """This class fetches JSON schemas from AWS AppConfig
+
+        Parameters
+        ----------
+        environment: str
+            Appconfig environment, e.g. 'dev/test' etc.
+        application: str
+            AppConfig application name, e.g. 'powertools'
+        name: str
+            AppConfig configuration name e.g. `my_conf`
+        cache_seconds: int
+            cache expiration time, how often to call AppConfig to fetch latest configuration
+        sdk_config: Optional[Config]
+            Botocore Config object to pass during client initialization
+        envelope : str
+            JMESPath expression to pluck feature flags data from config
+        jmespath_options : Dict
+            Alternative JMESPath options to be included when filtering expr
+        """
+        super().__init__()
+        self.environment = environment
+        self.application = application
+        self.name = name
+        self.cache_seconds = cache_seconds
+        self.config = sdk_config
+        self.envelope = envelope
+        self.jmespath_options = jmespath_options
+        self._conf_store = AppConfigProvider(environment=environment, application=application, config=sdk_config)
+
+    def get_configuration(self) -> Dict[str, Any]:
+        """Fetch feature schema configuration from AWS AppConfig
+
+        Raises
+        ------
+        ConfigurationStoreError
+            Any validation error or AppConfig error that can occur
+
+        Returns
+        -------
+        Dict[str, Any]
+            parsed JSON dictionary
+        """
+        try:
+            # parse result conf as JSON, keep in cache for self.max_age seconds
+            config = cast(
+                dict,
+                self._conf_store.get(
+                    name=self.name,
+                    transform=TRANSFORM_TYPE,
+                    max_age=self.cache_seconds,
+                ),
+            )
+
+            if self.envelope:
+                config = jmespath_utils.unwrap_event_from_envelope(
+                    data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
+                )
+
+            return config
+        except (GetParameterError, TransformParameterError) as exc:
+            err_msg = traceback.format_exc()
+            if "AccessDenied" in err_msg:
+                raise StoreClientError(err_msg) from exc
+            raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc

aws_lambda_powertools/utilities/feature_flags/base.py

@@ -0,0 +1,50 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+
+class StoreProvider(ABC):
+    @abstractmethod
+    def get_configuration(self) -> Dict[str, Any]:
+        """Get configuration from any store and return the parsed JSON dictionary
+
+        Raises
+        ------
+        ConfigurationStoreError
+            Any error that can occur during schema fetch or JSON parse
+
+        Returns
+        -------
+        Dict[str, Any]
+            parsed JSON dictionary
+
+            **Example**
+
+            ```python
+            {
+                "premium_features": {
+                    "default": False,
+                    "rules": {
+                        "customer tier equals premium": {
+                            "when_match": True,
+                            "conditions": [
+                                {
+                                    "action": "EQUALS",
+                                    "key": "tier",
+                                    "value": "premium",
+                                }
+                            ],
+                        }
+                    },
+                },
+                "feature_two": {
+                    "default": False
+                }
+            }
+        """
+        return NotImplemented  # pragma: no cover
+
+
+class BaseValidator(ABC):
+    @abstractmethod
+    def validate(self):
+        return NotImplemented  # pragma: no cover

aws_lambda_powertools/utilities/feature_flags/exceptions.py

@@ -0,0 +1,13 @@
+class ConfigurationStoreError(Exception):
+    """When a configuration store raises an exception on config retrieval or parsing"""
+
+
+class SchemaValidationError(Exception):
+    """When feature flag schema fails validation"""
+
+
+class StoreClientError(Exception):
+    """When a store raises an exception that should be propagated to the client to fix
+
+    For example, Access Denied errors when the client doesn't permissions to fetch config
+    """