Merge branch 'develop' of https://github.com/awslabs/aws-lambda-powertools-python into develop

* 'develop' of https://github.com/awslabs/aws-lambda-powertools-python: feat(feature-flags): Bring your own logger for debug (aws-powertools#709)

Author: heitorlessa

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -1,17 +1,16 @@
 import logging
 import traceback
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional, Union, cast
 
 from botocore.config import Config
 
 from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
 
+from ... import Logger
 from ...shared import jmespath_utils
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError, StoreClientError
 
-logger = logging.getLogger(__name__)
-
 TRANSFORM_TYPE = "json"
 
 
@@ -25,6 +24,7 @@ def __init__(
  sdk_config: Optional[Config] = None,
  envelope: Optional[str] = "",
  jmespath_options: Optional[Dict] = None,
+ logger: Optional[Union[logging.Logger, Logger]] = None,
  ):
  """This class fetches JSON schemas from AWS AppConfig
 
@@ -44,8 +44,11 @@ def __init__(
  JMESPath expression to pluck feature flags data from config
  jmespath_options : Optional[Dict]
  Alternative JMESPath options to be included when filtering expr
+ logger: A logging object
+ Used to log messages. If None is supplied, one will be created.
  """
  super().__init__()
+ self.logger = logger or logging.getLogger(__name__)
  self.environment = environment
  self.application = application
  self.name = name
@@ -60,6 +63,9 @@ 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
+ self.logger.debug(
+ "Fetching configuration from the store", extra={"param_name": self.name, "max_age": self.cache_seconds}
+ )
  return cast(
  dict,
  self._conf_store.get(
@@ -93,6 +99,7 @@ def get_configuration(self) -> Dict[str, Any]:
  config = self.get_raw_configuration
 
  if self.envelope:
+ self.logger.debug("Envelope enabled; extracting data from config", extra={"envelope": self.envelope})
  config = jmespath_utils.extract_data_from_envelope(
  data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
  )

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -1,15 +1,14 @@
 import logging
-from typing import Any, Dict, List, Optional, cast
+from typing import Any, Dict, List, Optional, Union, cast
 
+from ... import Logger
 from . import schema
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError
 
-logger = logging.getLogger(__name__)
-
 
 class FeatureFlags:
- def __init__(self, store: StoreProvider):
+ def __init__(self, store: StoreProvider, logger: Optional[Union[logging.Logger, Logger]] = None):
  """Evaluates whether feature flags should be enabled based on a given context.
 
  It uses the provided store to fetch feature flag rules before evaluating them.
@@ -35,11 +34,13 @@ def __init__(self, store: StoreProvider):
  ----------
  store: StoreProvider
  Store to use to fetch feature flag schema configuration.
+ logger: A logging object
+ Used to log messages. If None is supplied, one will be created.
  """
  self.store = store
+ self.logger = logger or logging.getLogger(__name__)
 
- @staticmethod
- def _match_by_action(action: str, condition_value: Any, context_value: Any) -> bool:
+ def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
  if not context_value:
  return False
  mapping_by_action = {
@@ -58,7 +59,7 @@ def _match_by_action(action: str, condition_value: Any, context_value: Any) -> b
  func = mapping_by_action.get(action, lambda a, b: False)
  return func(context_value, condition_value)
  except Exception as exc:
- logger.debug(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 _evaluate_conditions(
@@ -69,7 +70,7 @@ def _evaluate_conditions(
  conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY))
 
  if not conditions:
- logger.debug(
+ self.logger.debug(
  f"rule did not match, no conditions to match, rule_name={rule_name}, rule_value={rule_match_value}, "
  f"name={feature_name} "
  )
@@ -81,13 +82,13 @@ def _evaluate_conditions(
  cond_value = condition.get(schema.CONDITION_VALUE)
 
  if not self._match_by_action(action=cond_action, condition_value=cond_value, context_value=context_value):
- logger.debug(
+ self.logger.debug(
  f"rule did not match action, rule_name={rule_name}, rule_value={rule_match_value}, "
  f"name={feature_name}, context_value={str(context_value)} "
  )
  return False # context doesn't match condition
 
- logger.debug(f"rule matched, rule_name={rule_name}, rule_value={rule_match_value}, name={feature_name}")
+ self.logger.debug(f"rule matched, rule_name={rule_name}, rule_value={rule_match_value}, name={feature_name}")
  return True
 
  def _evaluate_rules(
@@ -98,12 +99,16 @@ def _evaluate_rules(
  rule_match_value = rule.get(schema.RULE_MATCH_VALUE)
 
  # Context might contain PII data; do not log its value
- logger.debug(f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={feat_default}")
+ self.logger.debug(
+ f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={feat_default}"
+ )
  if self._evaluate_conditions(rule_name=rule_name, feature_name=feature_name, rule=rule, context=context):
  return bool(rule_match_value)
 
  # no rule matched, return default value of feature
- 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={feat_default}, name={feature_name}"
+ )
  return feat_default
  return False
 
@@ -150,7 +155,7 @@ def get_configuration(self) -> 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}")
+ self.logger.debug(f"Fetching schema from registered store, store={self.store}")
  config: Dict = self.store.get_configuration()
  validator = schema.SchemaValidator(schema=config)
  validator.validate()
@@ -194,21 +199,21 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
  try:
  features = self.get_configuration()
  except ConfigurationStoreError as err:
- logger.debug(f"Failed to fetch feature flags from store, returning default provided, reason={err}")
+ self.logger.debug(f"Failed to fetch feature flags from store, returning default provided, reason={err}")
  return default
 
  feature = features.get(name)
  if feature is None:
- logger.debug(f"Feature not found; returning default provided, name={name}, default={default}")
+ self.logger.debug(f"Feature not found; returning default provided, name={name}, default={default}")
  return default
 
  rules = feature.get(schema.RULES_KEY)
  feat_default = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
  if not rules:
- logger.debug(f"no rules found, returning feature default, name={name}, default={feat_default}")
+ self.logger.debug(f"no rules found, returning feature default, name={name}, default={feat_default}")
  return bool(feat_default)
 
- logger.debug(f"looking for rule match, name={name}, default={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)
 
  def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> List[str]:
@@ -245,20 +250,20 @@ def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> L
  try:
  features: Dict[str, Any] = self.get_configuration()
  except ConfigurationStoreError as err:
- logger.debug(f"Failed to fetch feature flags from store, returning empty list, reason={err}")
+ self.logger.debug(f"Failed to fetch feature flags from store, returning empty list, reason={err}")
  return features_enabled
 
- logger.debug("Evaluating all features")
+ self.logger.debug("Evaluating all features")
  for name, feature in features.items():
  rules = feature.get(schema.RULES_KEY, {})
  feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
  if feature_default_value and not rules:
- logger.debug(f"feature is enabled by default and has no defined rules, name={name}")
+ 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
  ):
- logger.debug(f"feature's calculated value is True, name={name}")
+ self.logger.debug(f"feature's calculated value is True, name={name}")
  features_enabled.append(name)
 
  return features_enabled

aws_lambda_powertools/utilities/feature_flags/schema.py

@@ -1,12 +1,11 @@
 import logging
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
+from ... import Logger
 from .base import BaseValidator
 from .exceptions import SchemaValidationError
 
-logger = logging.getLogger(__name__)
-
 RULES_KEY = "rules"
 FEATURE_DEFAULT_VAL_KEY = "default"
 CONDITIONS_KEY = "conditions"
@@ -111,11 +110,12 @@ class SchemaValidator(BaseValidator):
  ```
  """
 
- def __init__(self, schema: Dict[str, Any]):
+ def __init__(self, schema: Dict[str, Any], logger: Optional[Union[logging.Logger, Logger]] = None):
  self.schema = schema
+ self.logger = logger or logging.getLogger(__name__)
 
  def validate(self) -> None:
- logger.debug("Validating schema")
+ self.logger.debug("Validating schema")
  if not isinstance(self.schema, dict):
  raise SchemaValidationError(f"Features must be a dictionary, schema={str(self.schema)}")
 
@@ -126,12 +126,13 @@ def validate(self) -> None:
 class FeaturesValidator(BaseValidator):
  """Validates each feature and calls RulesValidator to validate its rules"""
 
- def __init__(self, schema: Dict):
+ def __init__(self, schema: Dict, logger: Optional[Union[logging.Logger, Logger]] = None):
  self.schema = schema
+ self.logger = logger or logging.getLogger(__name__)
 
  def validate(self):
  for name, feature in self.schema.items():
- logger.debug(f"Attempting to validate feature '{name}'")
+ self.logger.debug(f"Attempting to validate feature '{name}'")
  self.validate_feature(name, feature)
  rules = RulesValidator(feature=feature)
  rules.validate()
@@ -149,21 +150,22 @@ def validate_feature(name, feature):
 class RulesValidator(BaseValidator):
  """Validates each rule and calls ConditionsValidator to validate each rule's conditions"""
 
- def __init__(self, feature: Dict[str, Any]):
+ def __init__(self, feature: Dict[str, Any], 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__)
 
  def validate(self):
  if not self.rules:
- logger.debug("Rules are empty, ignoring validation")
+ self.logger.debug("Rules are empty, ignoring validation")
  return
 
  if not isinstance(self.rules, dict):
  raise SchemaValidationError(f"Feature rules must be a dictionary, feature={self.feature_name}")
 
  for rule_name, rule in self.rules.items():
- logger.debug(f"Attempting to validate rule '{rule_name}'")
+ self.logger.debug(f"Attempting to validate rule '{rule_name}'")
  self.validate_rule(rule=rule, rule_name=rule_name, feature_name=self.feature_name)
  conditions = ConditionsValidator(rule=rule, rule_name=rule_name)
  conditions.validate()
@@ -189,24 +191,25 @@ def validate_rule_default_value(rule: Dict, rule_name: str):
 
 
 class ConditionsValidator(BaseValidator):
- def __init__(self, rule: Dict[str, Any], rule_name: str):
+ def __init__(self, rule: Dict[str, Any], rule_name: str, logger: Optional[Union[logging.Logger, Logger]] = None):
  self.conditions: List[Dict[str, Any]] = rule.get(CONDITIONS_KEY, {})
  self.rule_name = rule_name
+ self.logger = logger or logging.getLogger(__name__)
 
  def validate(self):
  if not self.conditions or not isinstance(self.conditions, list):
  raise SchemaValidationError(f"Invalid condition, rule={self.rule_name}")
 
  for condition in self.conditions:
+ # Condition can contain PII data; do not log condition value
+ self.logger.debug(f"Attempting to validate condition for '{self.rule_name}'")
  self.validate_condition(rule_name=self.rule_name, condition=condition)
 
  @staticmethod
  def validate_condition(rule_name: str, condition: Dict[str, str]) -> None:
  if not condition or not isinstance(condition, dict):
  raise SchemaValidationError(f"Feature rule condition must be a dictionary, rule={rule_name}")
 
- # Condition can contain PII data; do not log condition value
- logger.debug(f"Attempting to validate condition for '{rule_name}'")
  ConditionsValidator.validate_condition_action(condition=condition, rule_name=rule_name)
  ConditionsValidator.validate_condition_key(condition=condition, rule_name=rule_name)
  ConditionsValidator.validate_condition_value(condition=condition, rule_name=rule_name)

docs/utilities/feature_flags.md

@@ -580,6 +580,7 @@ Parameter | Default | Description
 **max_age** | `5` | Number of seconds to cache feature flags configuration fetched from AWS AppConfig
 **sdk_config** | `None` | [Botocore Config object](https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html){target="_blank"}
 **jmespath_options** | `None` | For advanced use cases when you want to bring your own [JMESPath functions](https://github.com/jmespath/jmespath.py#custom-functions){target="_blank"}
+**logger** | `logging.Logger` | Logger to use for debug. You can optionally supply an instance of Powertools Logger.
 
 === "appconfig_store_example.py"