feat: add get_raw_configuration property in store; expose store

Author: heitorlessa

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -55,9 +55,31 @@ def __init__(
         self.jmespath_options = jmespath_options
         self._conf_store = AppConfigProvider(environment=environment, application=application, config=sdk_config)
 
+    @property
+    def get_raw_configuration(self) -> Dict[str, Any]:
+        """Fetch feature schema configuration from AWS AppConfig"""
+        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,
+                ),
+            )
+        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
+
     def get_configuration(self) -> Dict[str, Any]:
         """Fetch feature schema configuration from AWS AppConfig
 
+        If envelope is set, it'll extract and return feature flags from configuration,
+        otherwise it'll return the entire configuration fetched from AWS AppConfig.
+
         Raises
         ------
         ConfigurationStoreError
@@ -68,25 +90,11 @@ def get_configuration(self) -> Dict[str, Any]:
         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,
-                ),
-            )
+        config = self.get_raw_configuration
 
-            if self.envelope:
-                config = jmespath_utils.extract_data_from_envelope(
-                    data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
-                )
+        if self.envelope:
+            config = jmespath_utils.extract_data_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
+        return config

aws_lambda_powertools/utilities/feature_flags/base.py

@@ -3,10 +3,19 @@
 
 
 class StoreProvider(ABC):
+    @property
+    @abstractmethod
+    def get_raw_configuration(self) -> Dict[str, Any]:
+        """Get configuration from any store and return the parsed JSON dictionary"""
+        raise NotImplementedError()  # pragma: no cover
+
     @abstractmethod
     def get_configuration(self) -> Dict[str, Any]:
         """Get configuration from any store and return the parsed JSON dictionary
 
+        If envelope is set, it'll extract and return feature flags from configuration,
+        otherwise it'll return the entire configuration fetched from the store.
+
         Raises
         ------
         ConfigurationStoreError
@@ -42,10 +51,10 @@ def get_configuration(self) -> Dict[str, Any]:
         }
         ```
         """
-        return NotImplemented  # pragma: no cover
+        raise NotImplementedError()  # pragma: no cover
 
 
 class BaseValidator(ABC):
     @abstractmethod
     def validate(self):
-        return NotImplemented  # pragma: no cover
+        raise NotImplementedError()  # pragma: no cover

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, cast
 
 from . import schema
 from .base import StoreProvider
@@ -36,7 +36,7 @@ def __init__(self, store: StoreProvider):
         store: StoreProvider
             Store to use to fetch feature flag schema configuration.
         """
-        self._store = store
+        self.store = store
 
     @staticmethod
     def _match_by_action(action: str, condition_value: Any, context_value: Any) -> bool:
@@ -103,7 +103,7 @@ def _evaluate_rules(
             return feat_default
         return False
 
-    def get_configuration(self) -> Union[Dict[str, Dict], Dict]:
+    def get_configuration(self) -> Dict:
         """Get validated feature flag schema from configured store.
 
         Largely used to aid testing, since it's called by `evaluate` and `get_enabled_features` methods.
@@ -146,8 +146,8 @@ def get_configuration(self) -> Union[Dict[str, Dict], Dict]:
         ```
         """
         # parse result conf as JSON, keep in cache for max age defined in store
-        logger.debug(f"Fetching schema from registered store, store={self._store}")
-        config = self._store.get_configuration()
+        logger.debug(f"Fetching schema from registered store, store={self.store}")
+        config: Dict = self.store.get_configuration()
         validator = schema.SchemaValidator(schema=config)
         validator.validate()
 

tests/functional/feature_flags/test_feature_flags.py

@@ -587,3 +587,15 @@ def test_get_feature_toggle_propagates_access_denied_error(mocker, config):
     # THEN raise StoreClientError error
     with pytest.raises(StoreClientError, match="AccessDeniedException") as err:
         feature_flags.evaluate(name="Foo", default=False)
+
+
+def test_get_configuration_with_envelope_and_raw(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {"log_level": "INFO", "features": {"my_feature": {"default": expected_value}}}
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config, envelope="features")
+
+    features_config = feature_flags.get_configuration()
+    config = feature_flags.store.get_raw_configuration
+
+    assert "log_level" in config
+    assert "log_level" not in features_config