refactor(feature-flags): optimize UX and maintenance

aws_lambda_powertools/utilities/feature_toggles/__init__.py

@@ -1,16 +1,16 @@
 """Advanced feature toggles utility
 """
-from .appconfig_fetcher import AppConfigFetcher
-from .configuration_store import ConfigurationStore
+from .appconfig import AppConfigStore
+from .base import StoreProvider
 from .exceptions import ConfigurationError
+from .feature_flags import FeatureFlags
 from .schema import ACTION, SchemaValidator
-from .schema_fetcher import SchemaFetcher
 
 __all__ = [
     "ConfigurationError",
-    "ConfigurationStore",
+    "FeatureFlags",
     "ACTION",
     "SchemaValidator",
-    "AppConfigFetcher",
-    "SchemaFetcher",
+    "AppConfigStore",
+    "StoreProvider",
 ]

aws_lambda_powertools/utilities/feature_toggles/appconfig_fetcher.py → aws_lambda_powertools/utilities/feature_toggles/appconfig.py

@@ -5,21 +5,20 @@
 
 from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
 
+from .base import StoreProvider
 from .exceptions import ConfigurationError
-from .schema_fetcher import SchemaFetcher
 
 logger = logging.getLogger(__name__)
 
-
 TRANSFORM_TYPE = "json"
 
 
-class AppConfigFetcher(SchemaFetcher):
+class AppConfigStore(StoreProvider):
     def __init__(
         self,
         environment: str,
-        service: str,
-        configuration_name: str,
+        application: str,
+        name: str,
         cache_seconds: int,
         config: Optional[Config] = None,
     ):
@@ -28,19 +27,19 @@ def __init__(
         Parameters
         ----------
         environment: str
-            what appconfig environment to use 'dev/test' etc.
-        service: str
-            what service name to use from the supplied environment
-        configuration_name: str
-            what configuration to take from the environment & service combination
+            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
         config: Optional[Config]
             boto3 client configuration
         """
-        super().__init__(configuration_name, cache_seconds)
+        super().__init__(name, cache_seconds)
         self._logger = logger
-        self._conf_store = AppConfigProvider(environment=environment, application=service, config=config)
+        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
@@ -60,12 +59,10 @@ def get_json_configuration(self) -> Dict[str, Any]:
             return cast(
                 dict,
                 self._conf_store.get(
-                    name=self.configuration_name,
+                    name=self.name,
                     transform=TRANSFORM_TYPE,
                     max_age=self._cache_seconds,
                 ),
             )
         except (GetParameterError, TransformParameterError) as exc:
-            error_str = f"unable to get AWS AppConfig configuration file, exception={str(exc)}"
-            self._logger.error(error_str)
-            raise ConfigurationError(error_str)
+            raise ConfigurationError("Unable to get AWS AppConfig configuration file") from exc

aws_lambda_powertools/utilities/feature_toggles/schema_fetcher.py → aws_lambda_powertools/utilities/feature_toggles/base.py

@@ -2,14 +2,14 @@
 from typing import Any, Dict
 
 
-class SchemaFetcher(ABC):
+class StoreProvider(ABC):
     def __init__(self, configuration_name: str, cache_seconds: int):
-        self.configuration_name = configuration_name
+        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 service and return the parsed JSON dictionary
+        """Get configuration string from any configuration storing application and return the parsed JSON dictionary
 
         Raises
         ------

aws_lambda_powertools/utilities/feature_toggles/configuration_store.py → aws_lambda_powertools/utilities/feature_toggles/feature_flags.py

@@ -2,23 +2,23 @@
 from typing import Any, Dict, List, Optional, cast
 
 from . import schema
+from .base import StoreProvider
 from .exceptions import ConfigurationError
-from .schema_fetcher import SchemaFetcher
 
 logger = logging.getLogger(__name__)
 
 
-class ConfigurationStore:
-    def __init__(self, schema_fetcher: SchemaFetcher):
+class FeatureFlags:
+    def __init__(self, store: StoreProvider):
         """constructor
 
         Parameters
         ----------
-        schema_fetcher: SchemaFetcher
+        store: StoreProvider
             A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
         """
         self._logger = logger
-        self._schema_fetcher = schema_fetcher
+        self._store = store
         self._schema_validator = schema.SchemaValidator(self._logger)
 
     def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
@@ -35,16 +35,16 @@ def _match_by_action(self, action: str, condition_value: Any, context_value: Any
             func = mapping_by_action.get(action, lambda a, b: False)
             return func(context_value, condition_value)
         except Exception as exc:
-            self._logger.error(f"caught exception while matching action, action={action}, exception={str(exc)}")
+            self._logger.debug(f"caught exception while matching action: action={action}, exception={str(exc)}")
             return False
 
-    def _is_rule_matched(self, feature_name: str, rule: Dict[str, Any], rules_context: Dict[str, Any]) -> bool:
+    def _is_rule_matched(self, feature_name: str, rule: Dict[str, Any], context: Dict[str, Any]) -> bool:
         rule_name = rule.get(schema.RULE_NAME_KEY, "")
         rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE)
         conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY))
 
         for condition in conditions:
-            context_value = rules_context.get(str(condition.get(schema.CONDITION_KEY)))
+            context_value = context.get(str(condition.get(schema.CONDITION_KEY)))
             if not self._match_by_action(
                 condition.get(schema.CONDITION_ACTION, ""),
                 condition.get(schema.CONDITION_VALUE),
@@ -68,13 +68,13 @@ def _handle_rules(
         self,
         *,
         feature_name: str,
-        rules_context: Dict[str, Any],
+        context: Dict[str, Any],
         feature_default_value: bool,
         rules: List[Dict[str, Any]],
     ) -> bool:
         for rule in rules:
             rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE)
-            if self._is_rule_matched(feature_name, rule, rules_context):
+            if self._is_rule_matched(feature_name, rule, context):
                 return bool(rule_default_value)
             # no rule matched, return default value of feature
             logger.debug(
@@ -98,14 +98,12 @@ def get_configuration(self) -> Dict[str, Any]:
             parsed JSON dictionary
         """
         # parse result conf as JSON, keep in cache for self.max_age seconds
-        config = self._schema_fetcher.get_json_configuration()
+        config = self._store.get_json_configuration()
         # validate schema
         self._schema_validator.validate_json_schema(config)
         return config
 
-    def get_feature_toggle(
-        self, *, feature_name: str, rules_context: Optional[Dict[str, Any]] = None, value_if_missing: bool
-    ) -> bool:
+    def evaluate(self, *, feature_name: str, context: Optional[Dict[str, Any]] = None, default: bool) -> bool:
         """Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
 
         See below for explanation.
@@ -114,13 +112,12 @@ def get_feature_toggle(
         ----------
         feature_name: str
             feature name that you wish to fetch
-        rules_context: Optional[Dict[str, Any]]
+        context: Optional[Dict[str, Any]]
             dict of attributes that you would like to match the rules
             against, can be {'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'} etc.
-        value_if_missing: bool
-            this will be the returned value in case the feature toggle doesn't exist in
-            the schema or there has been an error while fetching the
-            configuration from appconfig
+        default: bool
+            default value if feature flag doesn't exist in the schema,
+            or there has been an error while fetching the configuration from appconfig
 
         Returns
         ------
@@ -132,27 +129,27 @@ def get_feature_toggle(
                the defined feature
             3. feature exists and a rule matches -> rule_default_value of rule is returned
         """
-        if rules_context is None:
-            rules_context = {}
+        if context is None:
+            context = {}
 
         try:
             toggles_dict: Dict[str, Any] = self.get_configuration()
         except ConfigurationError:
-            logger.error("unable to get feature toggles JSON, returning provided value_if_missing value")
-            return value_if_missing
+            logger.debug("Unable to get feature toggles JSON, returning provided default value")
+            return default
 
         feature: Dict[str, Dict] = toggles_dict.get(schema.FEATURES_KEY, {}).get(feature_name, None)
         if feature is None:
-            logger.warning(
-                f"feature does not appear in configuration, using provided value_if_missing, "
-                f"feature_name={feature_name}, value_if_missing={value_if_missing}"
+            logger.debug(
+                f"feature does not appear in configuration, using provided default, "
+                f"feature_name={feature_name}, default={default}"
             )
-            return value_if_missing
+            return default
 
         rules_list = feature.get(schema.RULES_KEY)
         feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
         if not rules_list:
-            # not rules but has a value
+            # no rules but value
             logger.debug(
                 f"no rules found, returning feature default value, feature_name={feature_name}, "
                 f"default_value={feature_default_value}"
@@ -164,18 +161,18 @@ def get_feature_toggle(
         )
         return self._handle_rules(
             feature_name=feature_name,
-            rules_context=rules_context,
+            context=context,
             feature_default_value=bool(feature_default_value),
             rules=cast(List, rules_list),
         )
 
-    def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, Any]] = None) -> List[str]:
+    def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> List[str]:
         """Get all enabled feature toggles while also taking into account rule_context
         (when a feature has defined rules)
 
         Parameters
         ----------
-        rules_context: Optional[Dict[str, Any]]
+        context: Optional[Dict[str, Any]]
             dict of attributes that you would like to match the rules
             against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc.
 
@@ -185,16 +182,17 @@ def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, A
             a list of all features name that are enabled by also taking into account
             rule_context (when a feature has defined rules)
         """
-        if rules_context is None:
-            rules_context = {}
+        if context is None:
+            context = {}
+
+        features_enabled: List[str] = []
 
         try:
             toggles_dict: Dict[str, Any] = self.get_configuration()
         except ConfigurationError:
-            logger.error("unable to get feature toggles JSON")
-            return []
+            logger.debug("unable to get feature toggles JSON")
+            return features_enabled
 
-        ret_list = []
         features: Dict[str, Any] = toggles_dict.get(schema.FEATURES_KEY, {})
         for feature_name, feature_dict_def in features.items():
             rules_list = feature_dict_def.get(schema.RULES_KEY, [])
@@ -203,14 +201,14 @@ def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, A
                 self._logger.debug(
                     f"feature is enabled by default and has no defined rules, feature_name={feature_name}"
                 )
-                ret_list.append(feature_name)
+                features_enabled.append(feature_name)
             elif self._handle_rules(
                 feature_name=feature_name,
-                rules_context=rules_context,
+                context=context,
                 feature_default_value=feature_default_value,
                 rules=rules_list,
             ):
                 self._logger.debug(f"feature's calculated value is True, feature_name={feature_name}")
-                ret_list.append(feature_name)
+                features_enabled.append(feature_name)
 
-        return ret_list
+        return features_enabled

aws_lambda_powertools/utilities/feature_toggles/schema.py

@@ -26,59 +26,69 @@ class SchemaValidator:
     def __init__(self, logger: Logger):
         self._logger = logger
 
-    def _raise_conf_exc(self, error_str: str) -> None:
-        self._logger.error(error_str)
-        raise ConfigurationError(error_str)
-
-    def _validate_condition(self, rule_name: str, condition: Dict[str, str]) -> None:
+    @staticmethod
+    def _validate_condition(rule_name: str, condition: Dict[str, str]) -> None:
         if not condition or not isinstance(condition, dict):
-            self._raise_conf_exc(f"invalid condition type, not a dictionary, rule_name={rule_name}")
+            raise ConfigurationError(f"invalid condition type, not a dictionary, rule_name={rule_name}")
+
         action = condition.get(CONDITION_ACTION, "")
         if action not in [ACTION.EQUALS.value, ACTION.STARTSWITH.value, ACTION.ENDSWITH.value, ACTION.CONTAINS.value]:
-            self._raise_conf_exc(f"invalid action value, rule_name={rule_name}, action={action}")
+            raise ConfigurationError(f"invalid action value, rule_name={rule_name}, action={action}")
+
         key = condition.get(CONDITION_KEY, "")
         if not key or not isinstance(key, str):
-            self._raise_conf_exc(f"invalid key value, key has to be a non empty string, rule_name={rule_name}")
+            raise ConfigurationError(f"Invalid key value, key has to be a non empty string, rule_name={rule_name}")
+
         value = condition.get(CONDITION_VALUE, "")
         if not value:
-            self._raise_conf_exc(f"missing condition value, rule_name={rule_name}")
+            raise ConfigurationError(f"Missing condition value, rule_name={rule_name}")
 
     def _validate_rule(self, feature_name: str, rule: Dict[str, Any]) -> None:
         if not rule or not isinstance(rule, dict):
-            self._raise_conf_exc(f"feature rule is not a dictionary, feature_name={feature_name}")
+            raise ConfigurationError(f"Feature rule is not a dictionary, feature_name={feature_name}")
+
         rule_name = rule.get(RULE_NAME_KEY)
         if not rule_name or rule_name is None or not isinstance(rule_name, str):
-            return self._raise_conf_exc(f"invalid rule_name, feature_name={feature_name}")
+            raise ConfigurationError(f"Invalid rule_name, feature_name={feature_name}")
+
         rule_default_value = rule.get(RULE_DEFAULT_VALUE)
         if rule_default_value is None or not isinstance(rule_default_value, bool):
-            self._raise_conf_exc(f"invalid rule_default_value, rule_name={rule_name}")
+            raise ConfigurationError(f"Invalid rule_default_value, rule_name={rule_name}")
+
         conditions = rule.get(CONDITIONS_KEY, {})
         if not conditions or not isinstance(conditions, list):
-            self._raise_conf_exc(f"invalid condition, rule_name={rule_name}")
+            raise ConfigurationError(f"Invalid condition, rule_name={rule_name}")
+
         # validate conditions
         for condition in conditions:
             self._validate_condition(rule_name, condition)
 
     def _validate_feature(self, feature_name: str, feature_dict_def: Dict[str, Any]) -> None:
         if not feature_dict_def or not isinstance(feature_dict_def, dict):
-            self._raise_conf_exc(f"invalid AWS AppConfig JSON schema detected, feature {feature_name} is invalid")
+            raise ConfigurationError(f"Invalid AWS AppConfig JSON schema detected, feature {feature_name} is invalid")
+
         feature_default_value = feature_dict_def.get(FEATURE_DEFAULT_VAL_KEY)
         if feature_default_value is None or not isinstance(feature_default_value, bool):
-            self._raise_conf_exc(f"missing feature_default_value for feature, feature_name={feature_name}")
+            raise ConfigurationError(f"Missing feature_default_value for feature, feature_name={feature_name}")
+
         # validate rules
         rules = feature_dict_def.get(RULES_KEY, [])
         if not rules:
             return
+
         if not isinstance(rules, list):
-            self._raise_conf_exc(f"feature rules is not a list, feature_name={feature_name}")
+            raise ConfigurationError(f"Feature rules is not a list, feature_name={feature_name}")
+
         for rule in rules:
             self._validate_rule(feature_name, rule)
 
     def validate_json_schema(self, schema: Dict[str, Any]) -> None:
         if not isinstance(schema, dict):
-            self._raise_conf_exc("invalid AWS AppConfig JSON schema detected, root schema is not a dictionary")
+            raise ConfigurationError("invalid AWS AppConfig JSON schema detected, root schema is not a dictionary")
+
         features_dict = schema.get(FEATURES_KEY)
         if not isinstance(features_dict, dict):
-            return self._raise_conf_exc("invalid AWS AppConfig JSON schema detected, missing features dictionary")
+            raise ConfigurationError("invalid AWS AppConfig JSON schema detected, missing features dictionary")
+
         for feature_name, feature_dict_def in features_dict.items():
             self._validate_feature(feature_name, feature_dict_def)