feat(feat-toggle): New simple feature toggles rule engine (WIP) (#494)

Main features:

* Define global boolean feature toggles
* Define boolean feature toggles per customer, email or any other key/value. Keys are always strings, values can be other valid json types. 
* Use get_configuration API to get the entire configuration dict. Basically, it's an easy way to get a JSON file, the same way the AppConfig utility did.
* get_all_enabled_feature_toggles - get a list of strings - names of boolean feature toggles that are True according to the input context, i.e. all the rules that matched/True by default.
* Current recommended default is to use AppConfig as the feature store but allows for extension with other services via the Schema Fetcher.

Before releasing to prod we should fix:
* Missing docstrings with examples on how to use it in public Classes and public methods
* Document and explain the rules mechanism and rule match flow.
* Review whether we have sufficient logger.debug coverage for future diagnostic
* Docs: Extract key features for getting started vs advanced
* Use mypy doc strings

aws_lambda_powertools/utilities/feature_toggles/__init__.py

@@ -0,0 +1,16 @@
+"""Advanced feature toggles utility
+"""
+from .appconfig_fetcher import AppConfigFetcher
+from .configuration_store import ConfigurationStore
+from .exceptions import ConfigurationException
+from .schema import ACTION, SchemaValidator
+from .schema_fetcher import SchemaFetcher
+
+__all__ = [
+ "ConfigurationException",
+ "ConfigurationStore",
+ "ACTION",
+ "SchemaValidator",
+ "AppConfigFetcher",
+ "SchemaFetcher",
+]

aws_lambda_powertools/utilities/feature_toggles/appconfig_fetcher.py

@@ -0,0 +1,57 @@
+import logging
+from typing import Any, Dict, Optional
+
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
+
+from .exceptions import ConfigurationException
+from .schema_fetcher import SchemaFetcher
+
+logger = logging.getLogger(__name__)
+
+
+TRANSFORM_TYPE = "json"
+
+
+class AppConfigFetcher(SchemaFetcher):
+ def __init__(
+ self,
+ environment: str,
+ service: str,
+ configuration_name: str,
+ cache_seconds: int,
+ config: Optional[Config] = None,
+ ):
+ """This class fetches JSON schemas from AWS AppConfig
+
+ Args:
+ 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
+ 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)
+ self._logger = logger
+ self._conf_store = AppConfigProvider(environment=environment, application=service, config=config)
+
+ def get_json_configuration(self) -> Dict[str, Any]:
+ """Get configuration string from AWs AppConfig and return the parsed JSON dictionary
+
+ Raises:
+ ConfigurationException: Any validation error or appconfig error that can occur
+
+ Returns:
+ Dict[str, Any]: parsed JSON dictionary
+ """
+ try:
+ return self._conf_store.get(
+ name=self.configuration_name,
+ transform=TRANSFORM_TYPE,
+ max_age=self._cache_seconds,
+ ) # parse result conf as JSON, keep in cache for self.max_age 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 ConfigurationException(error_str)

aws_lambda_powertools/utilities/feature_toggles/configuration_store.py

@@ -0,0 +1,191 @@
+import logging
+from typing import Any, Dict, List, Optional
+
+from . import schema
+from .exceptions import ConfigurationException
+from .schema_fetcher import SchemaFetcher
+
+logger = logging.getLogger(__name__)
+
+
+class ConfigurationStore:
+ def __init__(self, schema_fetcher: SchemaFetcher):
+ """constructor
+
+ Args:
+ schema_fetcher (SchemaFetcher): A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
+ """
+ self._logger = logger
+ self._schema_fetcher = schema_fetcher
+ self._schema_validator = schema.SchemaValidator(self._logger)
+
+ def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
+ if not context_value:
+ return False
+ mapping_by_action = {
+ schema.ACTION.EQUALS.value: lambda a, b: a == b,
+ schema.ACTION.STARTSWITH.value: lambda a, b: a.startswith(b),
+ schema.ACTION.ENDSWITH.value: lambda a, b: a.endswith(b),
+ schema.ACTION.CONTAINS.value: lambda a, b: a in b,
+ }
+
+ try:
+ 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)}")
+ return False
+
+ def _is_rule_matched(self, feature_name: str, rule: Dict[str, Any], rules_context: Dict[str, Any]) -> bool:
+ rule_name = rule.get(schema.RULE_NAME_KEY, "")
+ rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE)
+ conditions: Dict[str, str] = rule.get(schema.CONDITIONS_KEY)
+
+ for condition in conditions:
+ context_value = rules_context.get(condition.get(schema.CONDITION_KEY))
+ if not self._match_by_action(
+ condition.get(schema.CONDITION_ACTION),
+ condition.get(schema.CONDITION_VALUE),
+ context_value,
+ ):
+ logger.debug(
+ f"rule did not match action, rule_name={rule_name}, rule_default_value={rule_default_value}, feature_name={feature_name}, context_value={str(context_value)}" # noqa: E501
+ )
+ # context doesn't match condition
+ return False
+ # if we got here, all conditions match
+ logger.debug(
+ f"rule matched, rule_name={rule_name}, rule_default_value={rule_default_value}, feature_name={feature_name}" # noqa: E501
+ )
+ return True
+
+ def _handle_rules(
+ self,
+ *,
+ feature_name: str,
+ rules_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):
+ return rule_default_value
+ # no rule matched, return default value of feature
+ logger.debug(
+ f"no rule matched, returning default value of feature, feature_default_value={feature_default_value}, feature_name={feature_name}" # noqa: E501
+ )
+ return feature_default_value
+
+ def get_configuration(self) -> Dict[str, Any]:
+ """Get configuration string from AWs AppConfig and returned the parsed JSON dictionary
+
+ Raises:
+ ConfigurationException: Any validation error or appconfig error that can occur
+
+ Returns:
+ Dict[str, Any]: parsed JSON dictionary
+ """
+ schema: Dict[
+ str, Any
+ ] = (
+ self._schema_fetcher.get_json_configuration()
+ ) # parse result conf as JSON, keep in cache for self.max_age seconds
+ # validate schema
+ self._schema_validator.validate_json_schema(schema)
+ return schema
+
+ def get_feature_toggle(
+ self, *, feature_name: str, rules_context: Optional[Dict[str, Any]] = None, value_if_missing: bool
+ ) -> bool:
+ """get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
+ see below for explanation.
+
+ Args:
+ feature_name (str): feature name that you wish to fetch
+ rules_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
+
+ Returns:
+ bool: calculated feature toggle value. several possibilities:
+ 1. if the feature doesn't appear in the schema or there has been an error fetching the
+ configuration -> error/warning log would appear and value_if_missing is returned
+ 2. feature exists and has no rules or no rules have matched -> return feature_default_value of
+ the defined feature
+ 3. feature exists and a rule matches -> rule_default_value of rule is returned
+ """
+ if rules_context is None:
+ rules_context = {}
+
+ try:
+ toggles_dict: Dict[str, Any] = self.get_configuration()
+ except ConfigurationException:
+ logger.error("unable to get feature toggles JSON, returning provided value_if_missing value") # noqa: E501
+ return value_if_missing
+
+ 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, feature_name={feature_name}, value_if_missing={value_if_missing}" # noqa: E501
+ )
+ return value_if_missing
+
+ 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
+ logger.debug(
+ f"no rules found, returning feature default value, feature_name={feature_name}, default_value={feature_default_value}" # noqa: E501
+ )
+ return feature_default_value
+ # look for first rule match
+ logger.debug(
+ f"looking for rule match, feature_name={feature_name}, feature_default_value={feature_default_value}"
+ ) # noqa: E501
+ return self._handle_rules(
+ feature_name=feature_name,
+ rules_context=rules_context,
+ feature_default_value=feature_default_value,
+ rules=rules_list,
+ )
+
+ def get_all_enabled_feature_toggles(self, *, rules_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)
+
+ Args:
+ rules_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.
+
+ Returns:
+ List[str]: 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 = {}
+ try:
+ toggles_dict: Dict[str, Any] = self.get_configuration()
+ except ConfigurationException:
+ logger.error("unable to get feature toggles JSON") # noqa: E501
+ return []
+ 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, [])
+ feature_default_value = feature_dict_def.get(schema.FEATURE_DEFAULT_VAL_KEY)
+ if feature_default_value and not rules_list:
+ self._logger.debug(
+ f"feature is enabled by default and has no defined rules, feature_name={feature_name}"
+ )
+ ret_list.append(feature_name)
+ elif self._handle_rules(
+ feature_name=feature_name,
+ rules_context=rules_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)
+ return ret_list

aws_lambda_powertools/utilities/feature_toggles/exceptions.py

@@ -0,0 +1,2 @@
+class ConfigurationException(Exception):
+ """When a a configuration store raises an exception on config retrieval or parsing"""

aws_lambda_powertools/utilities/feature_toggles/schema.py

@@ -0,0 +1,83 @@
+from enum import Enum
+from typing import Any, Dict
+
+from .exceptions import ConfigurationException
+
+FEATURES_KEY = "features"
+RULES_KEY = "rules"
+FEATURE_DEFAULT_VAL_KEY = "feature_default_value"
+CONDITIONS_KEY = "conditions"
+RULE_NAME_KEY = "rule_name"
+RULE_DEFAULT_VALUE = "value_when_applies"
+CONDITION_KEY = "key"
+CONDITION_VALUE = "value"
+CONDITION_ACTION = "action"
+
+
+class ACTION(str, Enum):
+ EQUALS = "EQUALS"
+ STARTSWITH = "STARTSWITH"
+ ENDSWITH = "ENDSWITH"
+ CONTAINS = "CONTAINS"
+
+
+class SchemaValidator:
+ def __init__(self, logger: object):
+ self._logger = logger
+
+ def _raise_conf_exc(self, error_str: str) -> None:
+ self._logger.error(error_str)
+ raise ConfigurationException(error_str)
+
+ def _validate_condition(self, 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}")
+ 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}")
+ 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}")
+ value = condition.get(CONDITION_VALUE, "")
+ if not value:
+ self._raise_conf_exc(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}")
+ rule_name = rule.get(RULE_NAME_KEY)
+ if not rule_name or rule_name is None or not isinstance(rule_name, str):
+ self._raise_conf_exc(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}")
+ conditions = rule.get(CONDITIONS_KEY, {})
+ if not conditions or not isinstance(conditions, list):
+ self._raise_conf_exc(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")
+ 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}")
+ # 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}")
+ 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")
+ features_dict: Dict = schema.get(FEATURES_KEY)
+ if not isinstance(features_dict, dict):
+ self._raise_conf_exc("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)

aws_lambda_powertools/utilities/feature_toggles/schema_fetcher.py

@@ -0,0 +1,20 @@
+from abc import ABC, abstractclassmethod
+from typing import Any, Dict
+
+
+class SchemaFetcher(ABC):
+ def __init__(self, configuration_name: str, cache_seconds: int):
+ self.configuration_name = configuration_name
+ self._cache_seconds = cache_seconds
+
+ @abstractclassmethod
+ def get_json_configuration(self) -> Dict[str, Any]:
+ """Get configuration string from any configuration storing service and return the parsed JSON dictionary
+
+ Raises:
+ ConfigurationException: Any error that can occur during schema fetch or JSON parse
+
+ Returns:
+ Dict[str, Any]: parsed JSON dictionary
+ """
+ return None

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -149,7 +149,7 @@ def get_app_config(
  >>> print(value)
  My configuration value
 
- **Retrieves a confiugration value and decodes it using a JSON decoder**
+ **Retrieves a configuration value and decodes it using a JSON decoder**
 
  >>> from aws_lambda_powertools.utilities.parameters import get_parameter
  >>>