feat(feature_flags): support beyond boolean values (JSON values)

aws_lambda_powertools/shared/types.py

@@ -1,3 +1,5 @@
-from typing import Any, Callable, TypeVar
+from typing import Any, Callable, Dict, List, TypeVar, Union
 
 AnyCallableT = TypeVar("AnyCallableT", bound=Callable[..., Any])  # noqa: VNE001
+# JSON primitives only, mypy doesn't support recursive tho
+JSONType = Union[str, int, float, bool, None, Dict[str, Any], List[Any]]

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -2,6 +2,7 @@
 from typing import Any, Dict, List, Optional, Union, cast
 
 from ... import Logger
+from ...shared.types import JSONType
 from . import schema
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError
@@ -97,21 +98,30 @@ def _evaluate_conditions(
         return True
 
     def _evaluate_rules(
-        self, *, feature_name: str, context: Dict[str, Any], feat_default: bool, rules: Dict[str, Any]
+        self,
+        *,
+        feature_name: str,
+        context: Dict[str, Any],
+        feat_default: Any,
+        rules: Dict[str, Any],
+        boolean_feature: bool,
     ) -> bool:
         """Evaluates whether context matches rules and conditions, otherwise return feature default"""
         for rule_name, rule in rules.items():
             rule_match_value = rule.get(schema.RULE_MATCH_VALUE)
 
             # Context might contain PII data; do not log its value
             self.logger.debug(
-                f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={feat_default}"
+                f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={str(feat_default)}, boolean_feature={boolean_feature}"  # noqa: E501
             )
             if self._evaluate_conditions(rule_name=rule_name, feature_name=feature_name, rule=rule, context=context):
-                return bool(rule_match_value)
+                # Maintenance: Revisit before going GA.
+                return bool(rule_match_value) if boolean_feature else rule_match_value
 
         # no rule matched, return default value of feature
-        self.logger.debug(f"no rule matched, returning feature default, default={feat_default}, name={feature_name}")
+        self.logger.debug(
+            f"no rule matched, returning feature default, default={str(feat_default)}, name={feature_name}, boolean_feature={boolean_feature}"  # noqa: E501
+        )
         return feat_default
 
     def get_configuration(self) -> Dict:
@@ -164,7 +174,7 @@ def get_configuration(self) -> Dict:
 
         return config
 
-    def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, default: bool) -> bool:
+    def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, default: JSONType) -> JSONType:
         """Evaluate whether a feature flag should be enabled according to stored schema and input context
 
         **Logic when evaluating a feature flag**
@@ -181,14 +191,15 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
             Attributes that should be evaluated against the stored schema.
 
             for example: `{"tenant_id": "X", "username": "Y", "region": "Z"}`
-        default: bool
+        default: JSONType
             default value if feature flag doesn't exist in the schema,
             or there has been an error when fetching the configuration from the store
+            Can be boolean or any JSON values for non-boolean features.
 
         Returns
         ------
-        bool
-            whether feature should be enabled or not
+        JSONType
+            whether feature should be enabled (bool flags) or JSON value when non-bool feature matches
 
         Raises
         ------
@@ -211,12 +222,27 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
 
         rules = feature.get(schema.RULES_KEY)
         feat_default = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
+        # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
+        # for non-boolean flags. It'll need minor implementation changes, docs changes, and maybe refactor
+        # get_enabled_features. We can minimize breaking change, despite Beta label, by having a new
+        # method `get_matching_features` returning Dict[feature_name, feature_value]
+        boolean_feature = feature.get(
+            schema.FEATURE_DEFAULT_VAL_TYPE_KEY, True
+        )  # backwards compatability ,assume feature flag
         if not rules:
-            self.logger.debug(f"no rules found, returning feature default, name={name}, default={feat_default}")
-            return bool(feat_default)
+            self.logger.debug(
+                f"no rules found, returning feature default, name={name}, default={str(feat_default)}, boolean_feature={boolean_feature}"  # noqa: E501
+            )
+            # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
+            # for non-boolean flags.
+            return bool(feat_default) if boolean_feature else feat_default
 
-        self.logger.debug(f"looking for rule match, name={name}, default={feat_default}")
-        return self._evaluate_rules(feature_name=name, context=context, feat_default=bool(feat_default), rules=rules)
+        self.logger.debug(
+            f"looking for rule match, name={name}, default={str(feat_default)}, boolean_feature={boolean_feature}"  # noqa: E501
+        )
+        return self._evaluate_rules(
+            feature_name=name, context=context, feat_default=feat_default, rules=rules, boolean_feature=boolean_feature
+        )
 
     def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> List[str]:
         """Get all enabled feature flags while also taking into account context
@@ -259,11 +285,19 @@ def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> L
         for name, feature in features.items():
             rules = feature.get(schema.RULES_KEY, {})
             feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
+            boolean_feature = feature.get(
+                schema.FEATURE_DEFAULT_VAL_TYPE_KEY, True
+            )  # backwards compatability ,assume feature flag
+
             if feature_default_value and not rules:
                 self.logger.debug(f"feature is enabled by default and has no defined rules, name={name}")
                 features_enabled.append(name)
             elif self._evaluate_rules(
-                feature_name=name, context=context, feat_default=feature_default_value, rules=rules
+                feature_name=name,
+                context=context,
+                feat_default=feature_default_value,
+                rules=rules,
+                boolean_feature=boolean_feature,
             ):
                 self.logger.debug(f"feature's calculated value is True, name={name}")
                 features_enabled.append(name)

aws_lambda_powertools/utilities/feature_flags/schema.py

@@ -13,6 +13,7 @@
 CONDITION_KEY = "key"
 CONDITION_VALUE = "value"
 CONDITION_ACTION = "action"
+FEATURE_DEFAULT_VAL_TYPE_KEY = "boolean_type"
 
 
 class RuleAction(str, Enum):
@@ -48,13 +49,21 @@ class SchemaValidator(BaseValidator):
     A dictionary containing default value and rules for matching.
     The value MUST be an object and MIGHT contain the following members:
 
-    * **default**: `bool`. Defines default feature value. This MUST be present
+    * **default**: `Union[bool, JSONType]`. Defines default feature value. This MUST be present
+    * **boolean_type**: bool. Defines whether feature has non-boolean value (`JSONType`). This MIGHT be present
     * **rules**: `Dict[str, Dict]`. Rules object. This MIGHT be present
 
-    ```python
+    `JSONType` being any JSON primitive value: `Union[str, int, float, bool, None, Dict[str, Any], List[Any]]`
+
+    ```json
     {
         "my_feature": {
-            "default": True,
+            "default": true,
+            "rules": {}
+        },
+        "my_non_boolean_feature": {
+            "default": {"group": "read-only"},
+            "boolean_type": false,
             "rules": {}
         }
     }
@@ -65,16 +74,26 @@ class SchemaValidator(BaseValidator):
     A dictionary with each rule and their conditions that a feature might have.
     The value MIGHT be present, and when defined it MUST contain the following members:
 
-    * **when_match**: `bool`. Defines value to return when context matches conditions
+    * **when_match**: `Union[bool, JSONType]`. Defines value to return when context matches conditions
     * **conditions**: `List[Dict]`. Conditions object. This MUST be present
 
-    ```python
+    ```json
     {
         "my_feature": {
-            "default": True,
+            "default": true,
+            "rules": {
+                "tenant id equals 345345435": {
+                    "when_match": false,
+                    "conditions": []
+                }
+            }
+        },
+        "my_non_boolean_feature": {
+            "default": {"group": "read-only"},
+            "boolean_type": false,
             "rules": {
                 "tenant id equals 345345435": {
-                    "when_match": False,
+                    "when_match": {"group": "admin"},
                     "conditions": []
                 }
             }
@@ -94,13 +113,13 @@ class SchemaValidator(BaseValidator):
     * **key**: `str`. Key in given context to perform operation
     * **value**: `Any`. Value in given context that should match action operation.
 
-    ```python
+    ```json
     {
         "my_feature": {
-            "default": True,
+            "default": true,
             "rules": {
                 "tenant id equals 345345435": {
-                    "when_match": False,
+                    "when_match": false,
                     "conditions": [
                         {
                             "action": "EQUALS",
@@ -138,28 +157,38 @@ def __init__(self, schema: Dict, logger: Optional[Union[logging.Logger, Logger]]
     def validate(self):
         for name, feature in self.schema.items():
             self.logger.debug(f"Attempting to validate feature '{name}'")
-            self.validate_feature(name, feature)
-            rules = RulesValidator(feature=feature)
+            boolean_feature: bool = self.validate_feature(name, feature)
+            rules = RulesValidator(feature=feature, boolean_feature=boolean_feature)
             rules.validate()
 
+    # returns True in case the feature is a regular feature flag with a  boolean default value
     @staticmethod
-    def validate_feature(name, feature):
+    def validate_feature(name, feature) -> bool:
         if not feature or not isinstance(feature, dict):
             raise SchemaValidationError(f"Feature must be a non-empty dictionary, feature={name}")
 
-        default_value = feature.get(FEATURE_DEFAULT_VAL_KEY)
-        if default_value is None or not isinstance(default_value, bool):
+        default_value: Any = feature.get(FEATURE_DEFAULT_VAL_KEY)
+        boolean_feature: bool = feature.get(FEATURE_DEFAULT_VAL_TYPE_KEY, True)
+        # if feature is boolean_feature, default_value must be a boolean type.
+        # default_value must exist
+        # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
+        # for non-boolean flags.
+        if default_value is None or (not isinstance(default_value, bool) and boolean_feature):
             raise SchemaValidationError(f"feature 'default' boolean key must be present, feature={name}")
+        return boolean_feature
 
 
 class RulesValidator(BaseValidator):
     """Validates each rule and calls ConditionsValidator to validate each rule's conditions"""
 
-    def __init__(self, feature: Dict[str, Any], logger: Optional[Union[logging.Logger, Logger]] = None):
+    def __init__(
+        self, feature: Dict[str, Any], boolean_feature: bool, logger: Optional[Union[logging.Logger, Logger]] = None
+    ):
         self.feature = feature
         self.feature_name = next(iter(self.feature))
         self.rules: Optional[Dict] = self.feature.get(RULES_KEY)
         self.logger = logger or logging.getLogger(__name__)
+        self.boolean_feature = boolean_feature
 
     def validate(self):
         if not self.rules:
@@ -171,27 +200,29 @@ def validate(self):
 
         for rule_name, rule in self.rules.items():
             self.logger.debug(f"Attempting to validate rule '{rule_name}'")
-            self.validate_rule(rule=rule, rule_name=rule_name, feature_name=self.feature_name)
+            self.validate_rule(
+                rule=rule, rule_name=rule_name, feature_name=self.feature_name, boolean_feature=self.boolean_feature
+            )
             conditions = ConditionsValidator(rule=rule, rule_name=rule_name)
             conditions.validate()
 
     @staticmethod
-    def validate_rule(rule, rule_name, feature_name):
+    def validate_rule(rule: Dict, rule_name: str, feature_name: str, boolean_feature: bool = True):
         if not rule or not isinstance(rule, dict):
             raise SchemaValidationError(f"Feature rule must be a dictionary, feature={feature_name}")
 
         RulesValidator.validate_rule_name(rule_name=rule_name, feature_name=feature_name)
-        RulesValidator.validate_rule_default_value(rule=rule, rule_name=rule_name)
+        RulesValidator.validate_rule_default_value(rule=rule, rule_name=rule_name, boolean_feature=boolean_feature)
 
     @staticmethod
     def validate_rule_name(rule_name: str, feature_name: str):
         if not rule_name or not isinstance(rule_name, str):
             raise SchemaValidationError(f"Rule name key must have a non-empty string, feature={feature_name}")
 
     @staticmethod
-    def validate_rule_default_value(rule: Dict, rule_name: str):
+    def validate_rule_default_value(rule: Dict, rule_name: str, boolean_feature: bool):
         rule_default_value = rule.get(RULE_MATCH_VALUE)
-        if not isinstance(rule_default_value, bool):
+        if boolean_feature and not isinstance(rule_default_value, bool):
             raise SchemaValidationError(f"'rule_default_value' key must have be bool, rule={rule_name}")