feat: support JMESPath envelope/options to extract features

Author: heitorlessa

aws_lambda_powertools/shared/jmespath_functions.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/shared/jmespath_utils.py

@@ -5,6 +5,7 @@
 
 from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
 
+from ...shared import jmespath_utils
 from .base import StoreProvider
 from .exceptions import ConfigurationError
 
@@ -21,6 +22,8 @@ def __init__(
         name: str,
         cache_seconds: int,
         config: Optional[Config] = None,
+        envelope: str = "",
+        jmespath_options: Optional[Dict] = None,
     ):
         """This class fetches JSON schemas from AWS AppConfig
 
@@ -36,18 +39,28 @@ def __init__(
             cache expiration time, how often to call AppConfig to fetch latest configuration
         config: Optional[Config]
             boto3 client configuration
+        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__(name, cache_seconds)
-        self._logger = logger
+        super().__init__()
+        self.environment = environment
+        self.application = application
+        self.name = name
+        self.cache_seconds = cache_seconds
+        self.config = config
+        self.envelope = envelope
+        self.jmespath_options = jmespath_options
         self._conf_store = AppConfigProvider(environment=environment, application=application, config=config)
 
     def get_json_configuration(self) -> Dict[str, Any]:
-        """Get configuration string from AWs AppConfig and return the parsed JSON dictionary
+        """Get configuration string from AWS AppConfig and return the parsed JSON dictionary
 
         Raises
         ------
         ConfigurationError
-            Any validation error or appconfig error that can occur
+            Any validation error or AppConfig error that can occur
 
         Returns
         -------
@@ -56,13 +69,17 @@ def get_json_configuration(self) -> Dict[str, Any]:
         """
         try:
             # parse result conf as JSON, keep in cache for self.max_age seconds
-            return cast(
-                dict,
-                self._conf_store.get(
-                    name=self.name,
-                    transform=TRANSFORM_TYPE,
-                    max_age=self._cache_seconds,
-                ),
+            config = 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 cast(dict, config)
         except (GetParameterError, TransformParameterError) as exc:
             raise ConfigurationError("Unable to get AWS AppConfig configuration file") from exc

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -3,10 +3,6 @@
 
 
 class StoreProvider(ABC):
-    def __init__(self, configuration_name: str, cache_seconds: int):
-        self.name = configuration_name
-        self._cache_seconds = cache_seconds
-
     @abstractmethod
     def get_json_configuration(self) -> Dict[str, Any]:
         """Get configuration string from any configuration storing application and return the parsed JSON dictionary

aws_lambda_powertools/utilities/feature_flags/base.py

@@ -14,7 +14,7 @@
 import jmespath
 
 from aws_lambda_powertools.shared.cache_dict import LRUDict
-from aws_lambda_powertools.shared.jmespath_functions import PowertoolsFunctions
+from aws_lambda_powertools.shared.jmespath_utils import PowertoolsFunctions
 from aws_lambda_powertools.shared.json_encoder import Encoder
 from aws_lambda_powertools.utilities.idempotency.config import IdempotencyConfig
 from aws_lambda_powertools.utilities.idempotency.exceptions import (

aws_lambda_powertools/utilities/idempotency/persistence/base.py

@@ -1,13 +1,9 @@
 import logging
-from typing import Any, Dict, Optional, Union
+from typing import Dict, Optional, Union
 
 import fastjsonschema  # type: ignore
-import jmespath
-from jmespath.exceptions import LexerError  # type: ignore
 
-from aws_lambda_powertools.shared.jmespath_functions import PowertoolsFunctions
-
-from .exceptions import InvalidEnvelopeExpressionError, InvalidSchemaFormatError, SchemaValidationError
+from .exceptions import InvalidSchemaFormatError, SchemaValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -39,31 +35,3 @@ def validate_data_against_schema(data: Union[Dict, str], schema: Dict, formats:
     except fastjsonschema.JsonSchemaException as e:
         message = f"Failed schema validation. Error: {e.message}, Path: {e.path}, Data: {e.value}"  # noqa: B306, E501
         raise SchemaValidationError(message)
-
-
-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/validation/base.py

@@ -2,7 +2,8 @@
 from typing import Any, Callable, Dict, Optional, Union
 
 from ...middleware_factory import lambda_handler_decorator
-from .base import unwrap_event_from_envelope, validate_data_against_schema
+from ...shared import jmespath_utils
+from .base import validate_data_against_schema
 
 logger = logging.getLogger(__name__)
 
@@ -16,7 +17,7 @@ def validator(
     inbound_formats: Optional[Dict] = None,
     outbound_schema: Optional[Dict] = None,
     outbound_formats: Optional[Dict] = None,
-    envelope: Optional[str] = None,
+    envelope: str = "",
     jmespath_options: Optional[Dict] = None,
 ) -> Any:
     """Lambda handler decorator to validate incoming/outbound data using a JSON Schema
@@ -116,7 +117,9 @@ def handler(event, context):
         When JMESPath expression to unwrap event is invalid
     """
     if envelope:
-        event = unwrap_event_from_envelope(data=event, envelope=envelope, jmespath_options=jmespath_options)
+        event = jmespath_utils.unwrap_event_from_envelope(
+            data=event, envelope=envelope, jmespath_options=jmespath_options
+        )
 
     if inbound_schema:
         logger.debug("Validating inbound event")
@@ -216,6 +219,8 @@ def handler(event, context):
         When JMESPath expression to unwrap event is invalid
     """
     if envelope:
-        event = unwrap_event_from_envelope(data=event, envelope=envelope, jmespath_options=jmespath_options)
+        event = jmespath_utils.unwrap_event_from_envelope(
+            data=event, envelope=envelope, jmespath_options=jmespath_options
+        )
 
     validate_data_against_schema(data=event, schema=schema, formats=formats)

aws_lambda_powertools/utilities/validation/validator.py

@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import pytest
 from botocore.config import Config
@@ -15,7 +15,9 @@ def config():
     return Config(region_name="us-east-1")
 
 
-def init_feature_flags(mocker, mock_schema: Dict, config: Config) -> FeatureFlags:
+def init_feature_flags(
+    mocker, mock_schema: Dict, config: Config, envelope: str = "", jmespath_options: Optional[Dict] = None
+) -> FeatureFlags:
     mocked_get_conf = mocker.patch("aws_lambda_powertools.utilities.parameters.AppConfigProvider.get")
     mocked_get_conf.return_value = mock_schema
 
@@ -25,6 +27,8 @@ def init_feature_flags(mocker, mock_schema: Dict, config: Config) -> FeatureFlag
         name="test_conf_name",
         cache_seconds=600,
         config=config,
+        envelope=envelope,
+        jmespath_options=jmespath_options,
     )
     feature_flags: FeatureFlags = FeatureFlags(store=app_conf_fetcher)
     return feature_flags
@@ -73,7 +77,7 @@ def test_toggles_rule_does_not_match(mocker, config):
 # you get the default value of False that was sent to the evaluate API
 def test_toggles_no_conditions_feature_does_not_exist(mocker, config):
     expected_value = False
-    mocked_app_config_schema = {"features": {"my_fake_feature": {"default": True}}}
+    mocked_app_config_schema = {"my_fake_feature": {"default": True}}
 
     feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
     toggle = feature_flags.evaluate(name="my_feature", context={}, default=expected_value)
@@ -448,3 +452,11 @@ def test_is_rule_matched_no_matches(mocker, config):
 
     # THEN return False
     assert result is False
+
+
+def test_features_jmespath_envelope(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {"features": {"my_feature": {"default": expected_value}}}
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config, envelope="features")
+    toggle = feature_flags.evaluate(name="my_feature", context={}, default=False)
+    assert toggle == expected_value

tests/functional/feature_toggles/test_feature_toggles.py

@@ -11,11 +11,11 @@
 from botocore.config import Config
 from jmespath import functions
 
+from aws_lambda_powertools.shared.jmespath_utils import unwrap_event_from_envelope
 from aws_lambda_powertools.shared.json_encoder import Encoder
 from aws_lambda_powertools.utilities.idempotency import DynamoDBPersistenceLayer
 from aws_lambda_powertools.utilities.idempotency.idempotency import IdempotencyConfig
 from aws_lambda_powertools.utilities.validation import envelopes
-from aws_lambda_powertools.utilities.validation.base import unwrap_event_from_envelope
 from tests.functional.utils import load_event
 
 TABLE_NAME = "TEST_TABLE"