Add logger to class inits.

Author: Gerald W. Lester

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -10,8 +10,6 @@
 from .base import StoreProvider
 from .exceptions import ConfigurationStoreError, StoreClientError
 
-logger = logging.getLogger(__name__)
-
 TRANSFORM_TYPE = "json"
 
 
@@ -25,6 +23,7 @@ def __init__(
         sdk_config: Optional[Config] = None,
         envelope: Optional[str] = "",
         jmespath_options: Optional[Dict] = None,
+        logger=None,
     ):
         """This class fetches JSON schemas from AWS AppConfig
 
@@ -46,6 +45,10 @@ def __init__(
             Alternative JMESPath options to be included when filtering expr
         """
         super().__init__()
+        if logger == None:
+            self.logger = logging.getLogger(__name__)
+        else:
+            self.logger = logger
         self.environment = environment
         self.application = application
         self.name = name

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -5,11 +5,9 @@
 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=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.
@@ -37,9 +35,12 @@ def __init__(self, store: StoreProvider):
             Store to use to fetch feature flag schema configuration.
         """
         self._store = store
+        if logger == None:
+            self.logger = logging.getLogger(__name__)
+        else:
+            self.logger = logger
 
-    @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 = {
@@ -54,7 +55,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.loggerdebug(f"caught exception while matching action: action={action}, exception={str(exc)}")
             return False
 
     def _evaluate_conditions(
@@ -65,7 +66,7 @@ def _evaluate_conditions(
         conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY))
 
         if not conditions:
-            logger.debug(
+            self.loggerdebug(
                 f"rule did not match, no conditions to match, rule_name={rule_name}, rule_value={rule_match_value}, "
                 f"name={feature_name} "
             )
@@ -77,13 +78,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.loggerdebug(
                     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.loggerdebug(f"rule matched, rule_name={rule_name}, rule_value={rule_match_value}, name={feature_name}")
         return True
 
     def _evaluate_rules(
@@ -94,12 +95,12 @@ 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.loggerdebug(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.loggerdebug(f"no rule matched, returning feature default, default={feat_default}, name={feature_name}")
             return feat_default
         return False
 
@@ -146,7 +147,7 @@ def get_configuration(self) -> Union[Dict[str, Dict], 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.loggerdebug(f"Fetching schema from registered store, store={self._store}")
         config = self._store.get_configuration()
         validator = schema.SchemaValidator(schema=config)
         validator.validate()
@@ -190,21 +191,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.loggerdebug(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.loggerdebug(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.loggerdebug(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.loggerdebug(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]:
@@ -241,20 +242,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.loggerdebug(f"Failed to fetch feature flags from store, returning empty list, reason={err}")
             return features_enabled
 
-        logger.debug("Evaluating all features")
+        self.loggerdebug("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.loggerdebug(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.loggerdebug(f"feature's calculated value is True, name={name}")
                 features_enabled.append(name)
 
         return features_enabled

aws_lambda_powertools/utilities/feature_flags/schema.py

@@ -5,8 +5,6 @@
 from .base import BaseValidator
 from .exceptions import SchemaValidationError
 
-logger = logging.getLogger(__name__)
-
 RULES_KEY = "rules"
 FEATURE_DEFAULT_VAL_KEY = "default"
 CONDITIONS_KEY = "conditions"
@@ -105,11 +103,15 @@ class SchemaValidator(BaseValidator):
     ```
     """
 
-    def __init__(self, schema: Dict[str, Any]):
+    def __init__(self, schema: Dict[str, Any], logger=None):
         self.schema = schema
+        if logger == None:
+            self.logger = logging.getLogger(__name__)
+        else:
+            self.logger = logger
 
     def validate(self) -> None:
-        logger.debug("Validating schema")
+        self.loggerdebug("Validating schema")
         if not isinstance(self.schema, dict):
             raise SchemaValidationError(f"Features must be a dictionary, schema={str(self.schema)}")
 
@@ -125,7 +127,7 @@ def __init__(self, schema: Dict):
 
     def validate(self):
         for name, feature in self.schema.items():
-            logger.debug(f"Attempting to validate feature '{name}'")
+            self.loggerdebug(f"Attempting to validate feature '{name}'")
             self.validate_feature(name, feature)
             rules = RulesValidator(feature=feature)
             rules.validate()
@@ -150,14 +152,14 @@ def __init__(self, feature: Dict[str, Any]):
 
     def validate(self):
         if not self.rules:
-            logger.debug("Rules are empty, ignoring validation")
+            self.loggerdebug("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.loggerdebug(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()
@@ -194,13 +196,12 @@ def validate(self):
         for condition in self.conditions:
             self.validate_condition(rule_name=self.rule_name, condition=condition)
 
-    @staticmethod
-    def validate_condition(rule_name: str, condition: Dict[str, str]) -> None:
+    def validate_condition(self,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}'")
+        self.loggerdebug(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)