lint check should pass

Author: FarhanAnjum-opti

optimizely/decision_service.py

@@ -479,7 +479,6 @@ def get_variation_for_feature(
           Decision namedtuple consisting of experiment and variation for the user.
     """
         return self.get_variations_for_feature_list(project_config, [feature], user_context, options)[0]
-        
 
     def validated_forced_decision(
         self,
@@ -543,15 +542,15 @@ def validated_forced_decision(
                 user_context.logger.info(user_has_forced_decision_but_invalid)
 
         return None, reasons
-    
+
     def get_variations_for_feature_list(
         self,
         project_config: ProjectConfig,
         features: list[entities.FeatureFlag],
         user_context: OptimizelyUserContext,
         options: Optional[Sequence[str]] = None
-    )->list[tuple[Decision, list[str]]]:
-        """ 
+    ) -> list[tuple[Decision, list[str]]]:
+        """
         Returns the list of experiment/variation the user is bucketed in for the given list of features.
         Args:
                 project_config: Instance of ProjectConfig.
@@ -563,24 +562,23 @@ def get_variations_for_feature_list(
             List of Decision namedtuple consisting of experiment and variation for the user.
         """
         decide_reasons: list[str] = []
-        
+
         if options:
             ignore_ups = OptimizelyDecideOption.IGNORE_USER_PROFILE_SERVICE in options
         else:
             ignore_ups = False
-        
-        
+
         user_profile_tracker: Optional[UserProfileTracker] = None
         if self.user_profile_service is not None and not ignore_ups:
             user_profile_tracker = UserProfileTracker(user_context.user_id, self.user_profile_service, self.logger)
             user_profile_tracker.load_user_profile(decide_reasons, None)
-            
+
         decisions = []
-        
+
         for feature in features:
             feature_reasons = decide_reasons.copy()
             experiment_decision_found = False  # Track if an experiment decision was made for the feature
-            
+
             # Check if the feature flag is under an experiment
             if feature.experimentIds:
                 for experiment_id in feature.experimentIds:
@@ -603,28 +601,32 @@ def get_variations_for_feature_list(
                             feature_reasons.extend(variation_reasons)
 
                         if decision_variation:
-                            self.logger.debug(f'User "{user_context.user_id}" bucketed into experiment "{experiment.key}" of feature "{feature.key}".')
+                            self.logger.debug(
+                                'User "{}" bucketed into experiment "{}" of feature "{}".'.format(
+                                    user_context.user_id, experiment.key, feature.key)
+                            )
                             decision = Decision(experiment, decision_variation, enums.DecisionSources.FEATURE_TEST)
                             decisions.append((decision, feature_reasons))
                             experiment_decision_found = True  # Mark that a decision was found
                             break  # Stop after the first successful experiment decision
-            
+
             # Only process rollout if no experiment decision was found
             if not experiment_decision_found:
-                rollout_decision, rollout_reasons = self.get_variation_for_rollout(project_config, feature, user_context)
+                rollout_decision, rollout_reasons = self.get_variation_for_rollout(project_config,
+                                                                                   feature,
+                                                                                   user_context)
                 if rollout_reasons:
                     feature_reasons.extend(rollout_reasons)
                 if rollout_decision:
-                    self.logger.debug(f'User "{user_context.user_id}" bucketed into rollout for feature "{feature.key}".')
+                    self.logger.debug(f'User "{user_context.user_id}" '
+                                      f'bucketed into rollout for feature "{feature.key}".')
                 else:
-                    self.logger.debug(f'User "{user_context.user_id}" not bucketed into any rollout for feature "{feature.key}".')
+                    self.logger.debug(f'User "{user_context.user_id}" '
+                                      f'not bucketed into any rollout for feature "{feature.key}".')
 
                 decisions.append((rollout_decision, feature_reasons))
-                    
+
         if self.user_profile_service is not None and user_profile_tracker is not None and ignore_ups is False:
             user_profile_tracker.save_user_profile()
-        
-        return decisions
 
-                
-                
+        return decisions

optimizely/optimizely.py

@@ -633,7 +633,10 @@ def get_variation(
 
         user_context = OptimizelyUserContext(self, self.logger, user_id, attributes, False)
         user_profile_tracker = user_profile.UserProfileTracker(user_id, self.user_profile_service, self.logger)
-        variation, _ = self.decision_service.get_variation(project_config, experiment, user_context, user_profile_tracker)
+        variation, _ = self.decision_service.get_variation(project_config,
+                                                           experiment,
+                                                           user_context,
+                                                           user_profile_tracker)
         if variation:
             variation_key = variation.key
 
@@ -1123,16 +1126,16 @@ def _decide(
 
         if OptimizelyDecideOption.ENABLED_FLAGS_ONLY in decide_options:
             decide_options.remove(OptimizelyDecideOption.ENABLED_FLAGS_ONLY)
-        
+
         decision = self._decide_for_keys(
             user_context,
             [key],
             decide_options,
             True
         )[key]
-        
+
         return decision
-        
+
     def _create_optimizely_decision(
             self,
             user_context: OptimizelyUserContext,
@@ -1147,23 +1150,26 @@ def _create_optimizely_decision(
         if flag_decision.variation is not None:
             if flag_decision.variation.featureEnabled:
                 feature_enabled = True
-    
+
         self.logger.info(f'Feature {flag_key} is enabled for user {user_id} {feature_enabled}"')
-        
+
         # Create Optimizely Decision Result.
         attributes = user_context.get_user_attributes()
         rule_key = flag_decision.experiment.key if flag_decision.experiment else None
         all_variables = {}
         decision_source = flag_decision.source
         decision_event_dispatched = False
-    
+
         feature_flag = project_config.feature_key_map.get(flag_key)
 
         # Send impression event if Decision came from a feature
         # test and decide options doesn't include disableDecisionEvent
         if OptimizelyDecideOption.DISABLE_DECISION_EVENT not in decide_options:
             if decision_source == DecisionSources.FEATURE_TEST or project_config.send_flag_decisions:
-                self._send_impression_event(project_config, flag_decision.experiment, flag_decision.variation, flag_key, rule_key or '',
+                self._send_impression_event(project_config,
+                                            flag_decision.experiment,
+                                            flag_decision.variation,
+                                            flag_key, rule_key or '',
                                             str(decision_source), feature_enabled,
                                             user_id, attributes)
 
@@ -1189,7 +1195,11 @@ def _create_optimizely_decision(
                 all_variables[variable_key] = actual_value
 
         should_include_reasons = OptimizelyDecideOption.INCLUDE_REASONS in decide_options
-        variation_key = flag_decision.variation.key if flag_decision is not None and flag_decision.variation is not None else None
+        variation_key = (
+            flag_decision.variation.key
+            if flag_decision is not None and flag_decision.variation is not None
+            else None
+        )
         # Send notification
         self.notification_center.send_notifications(
             enums.NotificationTypes.DECISION,
@@ -1212,7 +1222,6 @@ def _create_optimizely_decision(
                                   rule_key=rule_key, flag_key=flag_key,
                                   user_context=user_context, reasons=decision_reasons if should_include_reasons else []
                                   )
-            
 
     def _decide_all(
         self,
@@ -1282,7 +1291,7 @@ def _decide_for_keys(
             self.logger.debug('Provided decide options is not an array. Using default decide options.')
             merged_decide_options = self.default_decide_options
 
-        enabled_flags_only = OptimizelyDecideOption.ENABLED_FLAGS_ONLY in merged_decide_options
+        # enabled_flags_only = OptimizelyDecideOption.ENABLED_FLAGS_ONLY in merged_decide_options
 
         decisions: dict[str, OptimizelyDecision] = {}
         valid_keys = []
@@ -1292,11 +1301,11 @@ def _decide_for_keys(
         #     if enabled_flags_only and not decision.enabled:
         #         continue
         #     decisions[key] = decision
-            
+
         project_config = self.config_manager.get_config()
         flags_without_forced_decision: list[entities.FeatureFlag] = []
         flag_decisions: dict[str, Decision] = {}
-        
+
         if project_config is None:
             return decisions
         for key in keys:
@@ -1307,39 +1316,34 @@ def _decide_for_keys(
             valid_keys.append(key)
             decision_reasons: list[str] = []
             decision_reasons_dict[key] = decision_reasons
-            
+
             optimizely_decision_context = OptimizelyUserContext.OptimizelyDecisionContext(flag_key=key, rule_key=None)
             forced_decision_response = self.decision_service.validated_forced_decision(project_config,
                                                                                        optimizely_decision_context,
                                                                                        user_context)
             variation, decision_reasons = forced_decision_response
             decision_reasons_dict[key] += decision_reasons
-            
+
             if variation:
                 decision = Decision(None, variation, enums.DecisionSources.FEATURE_TEST)
                 flag_decisions[key] = decision
             else:
-                # Regular decision
-                # decision, decision_reasons = self.decision_service.get_variation_for_feature(project_config,
-                #                                                                             feature_flag,
-                #                                                                             user_context, decide_options)
                 flags_without_forced_decision.append(feature_flag)
 
-        
         decision_list = self.decision_service.get_variations_for_feature_list(
             project_config,
             flags_without_forced_decision,
             user_context,
             merged_decide_options
         )
-        
+
         for i in range(0, len(flags_without_forced_decision)):
             decision = decision_list[i][0]
             reasons = decision_list[i][1]
             flag_key = flags_without_forced_decision[i].key
             flag_decisions[flag_key] = decision
             decision_reasons_dict[flag_key] += reasons
-            
+
         print(decision_reasons_dict)
         for key in valid_keys:
             flag_decision = flag_decisions[key]
@@ -1352,10 +1356,11 @@ def _decide_for_keys(
                 merged_decide_options,
                 project_config
             )
-            
-            if (OptimizelyDecideOption.ENABLED_FLAGS_ONLY not in merged_decide_options) or (optimizely_decision.enabled):
+            enabled_flags_only_missing = OptimizelyDecideOption.ENABLED_FLAGS_ONLY not in merged_decide_options
+            is_enabled = optimizely_decision.enabled
+            if enabled_flags_only_missing or is_enabled:
                 decisions[key] = optimizely_decision
-        
+
         return decisions
 
     def _setup_odp(self, sdk_key: Optional[str]) -> None:

optimizely/user_profile.py

@@ -16,16 +16,13 @@
 from sys import version_info
 from . import logger as _logging
 from . import decision_service
-from .helpers import enums
 if version_info < (3, 8):
     from typing_extensions import Final
 else:
     from typing import Final, TYPE_CHECKING  # type: ignore
-    
+
     if TYPE_CHECKING:
         # prevent circular dependenacy by skipping import at runtime
-        from .project_config import ProjectConfig
-        from .logger import Logger
         from .entities import Experiment, Variation
         from .decision_service import Decision
         from optimizely.error_handler import BaseErrorHandler
@@ -106,18 +103,23 @@ def save(self, user_profile: dict[str, Any]) -> None:
     """
         pass
 
+
 class UserProfileTracker:
-    def __init__(self, user_id: str, user_profile_service: Optional[UserProfileService], logger:Optional[_logging.Logger] = None):
+    def __init__(self,
+                 user_id: str,
+                 user_profile_service: Optional[UserProfileService],
+                 logger: Optional[_logging.Logger] = None):
         self.user_id = user_id
         self.user_profile_service = user_profile_service
         self.logger = _logging.adapt_logger(logger or _logging.NoOpLogger())
         self.profile_updated = False
         self.user_profile = UserProfile(user_id, {})
-    
+
     def get_user_profile(self) -> UserProfile:
         return self.user_profile
 
-    def load_user_profile(self, reasons: Optional[list[str]]=[], error_handler: Optional[BaseErrorHandler]=None) -> None:
+    def load_user_profile(self, reasons: Optional[list[str]] = [],
+                          error_handler: Optional[BaseErrorHandler] = None) -> None:
         reasons = reasons if reasons else []
         try:
             user_profile = self.user_profile_service.lookup(self.user_id) if self.user_profile_service else None
@@ -128,7 +130,7 @@ def load_user_profile(self, reasons: Optional[list[str]]=[], error_handler: Opti
             else:
                 if 'user_id' in user_profile and 'experiment_bucket_map' in user_profile:
                     self.user_profile = UserProfile(
-                        user_profile['user_id'], 
+                        user_profile['user_id'],
                         user_profile['experiment_bucket_map']
                     )
                     self.logger.info("User profile loaded successfully.")
@@ -142,10 +144,10 @@ def load_user_profile(self, reasons: Optional[list[str]]=[], error_handler: Opti
             self.logger.exception(f'Unable to retrieve user profile for user "{self.user_id}"as lookup failed.')
             # Todo: add error handler
             # error_handler.handle_error()
-        
+
         if self.user_profile is None:
             self.user_profile = UserProfile(self.user_id, {})
-            
+
     def update_user_profile(self, experiment: Experiment, variation: Variation) -> None:
         if experiment.id in self.user_profile.experiment_bucket_map:
             decision = self.user_profile.experiment_bucket_map[experiment.id]
@@ -157,12 +159,10 @@ def update_user_profile(self, experiment: Experiment, variation: Variation) -> N
                 )
         else:
             decision = decision_service.Decision(experiment=None, variation=variation, source=None)
-         
+
         self.user_profile.experiment_bucket_map[experiment.id] = decision
         self.profile_updated = True
-        # self.logger.info(f'Updated variation "{variation.id}" of experiment "{experiment.id}" for user "{self.user_profile.user_id}".')
-        
-        
+
     def save_user_profile(self, error_handler: Optional[BaseErrorHandler] = None) -> None:
         if not self.profile_updated:
             return
@@ -171,5 +171,6 @@ def save_user_profile(self, error_handler: Optional[BaseErrorHandler] = None) ->
                 self.user_profile_service.save(self.user_profile.__dict__)
                 self.logger.info(f'Saved user profile of user "{self.user_profile.user_id}".')
         except Exception as exception:
-            self.logger.warning(f'Failed to save user profile of user "{self.user_profile.user_id}".')
+            self.logger.warning(f'Failed to save user profile of user "{self.user_profile.user_id}" '
+                                f'for exception:{exception}".')
             # error_handler.handle_error(exception)