refactor(feature-flags): add intersection tests; structure refinement (aws-powertools#3775)

Co-authored-by: Ruben Fonseca <fonseka@gmail.com>

Author: heitorlessa

aws_lambda_powertools/utilities/feature_flags/__init__.py

@@ -1,4 +1,5 @@
 """Advanced feature flags utility"""
+
 from .appconfig import AppConfigStore
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError

aws_lambda_powertools/utilities/feature_flags/comparators.py

@@ -1,10 +1,11 @@
+from __future__ import annotations
+
 from datetime import datetime, tzinfo
 from typing import Any, Dict, Optional
 
 from dateutil.tz import gettz
 
 from .schema import HOUR_MIN_SEPARATOR, ModuloRangeValues, TimeValues
-from .exceptions import SchemaValidationError
 
 
 def _get_now_from_timezone(timezone: Optional[tzinfo]) -> datetime:
@@ -85,41 +86,64 @@ def compare_modulo_range(context_value: int, condition_value: Dict) -> bool:
  return start <= context_value % base <= end
 
 
-def compare_any_in_list(key_list, value_list):
- if not (isinstance(key_list, list) and isinstance(value_list, list)):
- raise SchemaValidationError()
- 
- results = False
- for key in key_list:
- if key in value_list:
- results = True
- break
- 
- return results
+def compare_any_in_list(context_value: list, condition_value: list) -> bool:
+ """Comparator for ANY_IN_VALUE action
+
+ Parameters
+ ----------
+ context_value : list
+ user-defined context for flag evaluation
+ condition_value : list
+ schema value available for condition being evaluated
+
+ Returns
+ -------
+ bool
+ Whether any list item in context_value is available in condition_value
+ """
+ if not isinstance(context_value, list):
+ raise ValueError("Context provided must be a list. Unable to compare ANY_IN_VALUE action.")
+
+ return any(key in condition_value for key in context_value)
+
 
+def compare_all_in_list(context_value: list, condition_value: list) -> bool:
+ """Comparator for ALL_IN_VALUE action
 
-def compare_all_in_list(key_list, value_list):
- if not (isinstance(key_list, list) and isinstance(value_list, list)):
- raise SchemaValidationError()
+ Parameters
+ ----------
+ context_value : list
+ user-defined context for flag evaluation
+ condition_value : list
+ schema value available for condition being evaluated
 
- results = True
- for key in key_list:
- if key not in value_list:
- results = False
- break
+ Returns
+ -------
+ bool
+ Whether all list items in context_value are available in condition_value
+ """
+ if not isinstance(context_value, list):
+ raise ValueError("Context provided must be a list. Unable to compare ALL_IN_VALUE action.")
 
- return results
+ return all(key in condition_value for key in context_value)
 
 
-def compare_none_in_list(key_list, value_list):
- if not (isinstance(key_list, list) and isinstance(value_list, list)):
- raise SchemaValidationError()
+def compare_none_in_list(context_value: list, condition_value: list) -> bool:
+ """Comparator for NONE_IN_VALUE action
 
- results = True
- for key in key_list:
- if key in value_list:
- results = False
- break
+ Parameters
+ ----------
+ context_value : list
+ user-defined context for flag evaluation
+ condition_value : list
+ schema value available for condition being evaluated
 
- return results
+ Returns
+ -------
+ bool
+ Whether list items in context_value are **not** available in condition_value
+ """
+ if not isinstance(context_value, list):
+ raise ValueError("Context provided must be a list. Unable to compare NONE_IN_VALUE action.")
 
+ return all(key not in condition_value for key in context_value)

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -1,21 +1,52 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast
+
+from typing_extensions import ParamSpec
 
 from ... import Logger
 from ...shared.types import JSONType
 from . import schema
 from .base import StoreProvider
 from .comparators import (
+ compare_all_in_list,
+ compare_any_in_list,
  compare_datetime_range,
  compare_days_of_week,
  compare_modulo_range,
+ compare_none_in_list,
  compare_time_range,
- compare_all_in_list,
- compare_any_in_list,
- compare_none_in_list
 )
 from .exceptions import ConfigurationStoreError
 
+T = TypeVar("T")
+P = ParamSpec("P")
+
+RULE_ACTION_MAPPING = {
+ schema.RuleAction.EQUALS.value: lambda a, b: a == b,
+ schema.RuleAction.NOT_EQUALS.value: lambda a, b: a != b,
+ schema.RuleAction.KEY_GREATER_THAN_VALUE.value: lambda a, b: a > b,
+ schema.RuleAction.KEY_GREATER_THAN_OR_EQUAL_VALUE.value: lambda a, b: a >= b,
+ schema.RuleAction.KEY_LESS_THAN_VALUE.value: lambda a, b: a < b,
+ schema.RuleAction.KEY_LESS_THAN_OR_EQUAL_VALUE.value: lambda a, b: a <= b,
+ schema.RuleAction.STARTSWITH.value: lambda a, b: a.startswith(b),
+ schema.RuleAction.ENDSWITH.value: lambda a, b: a.endswith(b),
+ schema.RuleAction.IN.value: lambda a, b: a in b,
+ schema.RuleAction.NOT_IN.value: lambda a, b: a not in b,
+ schema.RuleAction.KEY_IN_VALUE.value: lambda a, b: a in b,
+ schema.RuleAction.KEY_NOT_IN_VALUE.value: lambda a, b: a not in b,
+ schema.RuleAction.VALUE_IN_KEY.value: lambda a, b: b in a,
+ schema.RuleAction.VALUE_NOT_IN_KEY.value: lambda a, b: b not in a,
+ schema.RuleAction.ALL_IN_VALUE.value: lambda a, b: compare_all_in_list(a, b),
+ schema.RuleAction.ANY_IN_VALUE.value: lambda a, b: compare_any_in_list(a, b),
+ schema.RuleAction.NONE_IN_VALUE.value: lambda a, b: compare_none_in_list(a, b),
+ schema.RuleAction.SCHEDULE_BETWEEN_TIME_RANGE.value: lambda a, b: compare_time_range(a, b),
+ schema.RuleAction.SCHEDULE_BETWEEN_DATETIME_RANGE.value: lambda a, b: compare_datetime_range(a, b),
+ schema.RuleAction.SCHEDULE_BETWEEN_DAYS_OF_WEEK.value: lambda a, b: compare_days_of_week(a, b),
+ schema.RuleAction.MODULO_RANGE.value: lambda a, b: compare_modulo_range(a, b),
+}
+
 
 class FeatureFlags:
  def __init__(self, store: StoreProvider, logger: Optional[Union[logging.Logger, Logger]] = None):
@@ -49,37 +80,20 @@ def __init__(self, store: StoreProvider, logger: Optional[Union[logging.Logger,
  """
  self.store = store
  self.logger = logger or logging.getLogger(__name__)
+ self._exception_handlers: dict[Exception, Callable] = {}
 
  def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
- mapping_by_action = {
- schema.RuleAction.EQUALS.value: lambda a, b: a == b,
- schema.RuleAction.NOT_EQUALS.value: lambda a, b: a != b,
- schema.RuleAction.KEY_GREATER_THAN_VALUE.value: lambda a, b: a > b,
- schema.RuleAction.KEY_GREATER_THAN_OR_EQUAL_VALUE.value: lambda a, b: a >= b,
- schema.RuleAction.KEY_LESS_THAN_VALUE.value: lambda a, b: a < b,
- schema.RuleAction.KEY_LESS_THAN_OR_EQUAL_VALUE.value: lambda a, b: a <= b,
- schema.RuleAction.STARTSWITH.value: lambda a, b: a.startswith(b),
- schema.RuleAction.ENDSWITH.value: lambda a, b: a.endswith(b),
- schema.RuleAction.IN.value: lambda a, b: a in b,
- schema.RuleAction.NOT_IN.value: lambda a, b: a not in b,
- schema.RuleAction.KEY_IN_VALUE.value: lambda a, b: a in b,
- schema.RuleAction.KEY_NOT_IN_VALUE.value: lambda a, b: a not in b,
- schema.RuleAction.VALUE_IN_KEY.value: lambda a, b: b in a,
- schema.RuleAction.VALUE_NOT_IN_KEY.value: lambda a, b: b not in a,
- schema.RuleAction.ALL_IN_VALUE.value: lambda a, b: compare_all_in_list(a, b),
- schema.RuleAction.ANY_IN_VALUE.value: lambda a, b: compare_any_in_list(a, b),
- schema.RuleAction.NONE_IN_VALUE.value: lambda a, b: compare_none_in_list(a, b),
- schema.RuleAction.SCHEDULE_BETWEEN_TIME_RANGE.value: lambda a, b: compare_time_range(a, b),
- schema.RuleAction.SCHEDULE_BETWEEN_DATETIME_RANGE.value: lambda a, b: compare_datetime_range(a, b),
- schema.RuleAction.SCHEDULE_BETWEEN_DAYS_OF_WEEK.value: lambda a, b: compare_days_of_week(a, b),
- schema.RuleAction.MODULO_RANGE.value: lambda a, b: compare_modulo_range(a, b),
- }
-
  try:
- func = mapping_by_action.get(action, lambda a, b: False)
+ func = RULE_ACTION_MAPPING.get(action, lambda a, b: False)
  return func(context_value, condition_value)
  except Exception as exc:
  self.logger.debug(f"caught exception while matching action: action={action}, exception={str(exc)}")
+
+ handler = self._lookup_exception_handler(exc)
+ if handler:
+ self.logger.debug("Exception handler found! Delegating response.")
+ return handler(exc)
+
  return False
 
  def _evaluate_conditions(
@@ -209,6 +223,22 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
  2. Feature exists but has either no rules or no match, return feature default value
  3. Feature doesn't exist in stored schema, encountered an error when fetching -> return default value provided
 
+ ┌────────────────────────┐ ┌────────────────────────┐ ┌────────────────────────┐
+ │ Feature flags │──────▶ Get Configuration ├───────▶ Evaluate rules │
+ └────────────────────────┘ │ │ │ │
+ │┌──────────────────────┐│ │┌──────────────────────┐│
+ ││ Fetch schema ││ ││ Match rule ││
+ │└───────────┬──────────┘│ │└───────────┬──────────┘│
+ │ │ │ │ │ │
+ │┌───────────▼──────────┐│ │┌───────────▼──────────┐│
+ ││ Cache schema ││ ││ Match condition ││
+ │└───────────┬──────────┘│ │└───────────┬──────────┘│
+ │ │ │ │ │ │
+ │┌───────────▼──────────┐│ │┌───────────▼──────────┐│
+ ││ Validate schema ││ ││ Match action ││
+ │└──────────────────────┘│ │└──────────────────────┘│
+ └────────────────────────┘ └────────────────────────┘
+
  Parameters
  ----------
  name: str
@@ -222,6 +252,31 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
  or there has been an error when fetching the configuration from the store
  Can be boolean or any JSON values for non-boolean features.
 
+
+ Examples
+ --------
+
+ ```python
+ from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags
+ from aws_lambda_powertools.utilities.typing import LambdaContext
+
+ app_config = AppConfigStore(environment="dev", application="product-catalogue", name="features")
+
+ feature_flags = FeatureFlags(store=app_config)
+
+
+ def lambda_handler(event: dict, context: LambdaContext):
+ # Get customer's tier from incoming request
+ ctx = {"tier": event.get("tier", "standard")}
+
+ # Evaluate whether customer's tier has access to premium features
+ # based on `has_premium_features` rules
+ has_premium_features: bool = feature_flags.evaluate(name="premium_features", context=ctx, default=False)
+ if has_premium_features:
+ # enable premium features
+ ...
+ ```
+
  Returns
  ------
  JSONType
@@ -335,3 +390,45 @@ def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> L
  features_enabled.append(name)
 
  return features_enabled
+
+ def validation_exception_handler(self, exc_class: Exception | list[Exception]):
+ """Registers function to handle unexpected validation exceptions when evaluating flags.
+
+ It does not override the function of a default flag value in case of network and IAM permissions.
+ For example, you won't be able to catch ConfigurationStoreError exception.
+
+ Parameters
+ ----------
+ exc_class : Exception | list[Exception]
+ One or more exceptions to catch
+
+ Examples
+ --------
+
+ ```python
+ feature_flags = FeatureFlags(store=app_config)
+
+ @feature_flags.validation_exception_handler(Exception) # any exception
+ def catch_exception(exc):
+ raise TypeError("re-raised") from exc
+ ```
+ """
+
+ def register_exception_handler(func: Callable[P, T]) -> Callable[P, T]:
+ if isinstance(exc_class, list):
+ for exp in exc_class:
+ self._exception_handlers[exp] = func
+ else:
+ self._exception_handlers[exc_class] = func
+
+ return func
+
+ return register_exception_handler
+
+ def _lookup_exception_handler(self, exc: BaseException) -> Callable | None:
+ # Use "Method Resolution Order" to allow for matching against a base class
+ # of an exception
+ for cls in type(exc).__mro__:
+ if cls in self._exception_handlers:
+ return self._exception_handlers[cls] # type: ignore[index] # index is correct
+ return None