new eval engine integrated into client

enver-bisevac · enver-bisevac · commit ef98a507d05f · 2022-04-25T10:47:58.000+02:00
diff --git a/featureflags/analytics.py b/featureflags/analytics.py
@@ -38,7 +38,7 @@
 @attr.s(auto_attribs=True)
 class AnalyticsEvent(object):
     target: Target
-    feature_config: FeatureConfig
+    flag_identifier: str
     variation: Variation
     count: int = 0
 
@@ -58,11 +58,11 @@ def __init__(self, config: Config, client: AuthenticatedClient,
         self._runner.daemon = True
         self._runner.start()
 
-    def enqueue(self, target: Target, feature_config: FeatureConfig,
+    def enqueue(self, target: Target, identifier: str,
                 variation: Variation):
         event: AnalyticsEvent = AnalyticsEvent(
             target=target,
-            feature_config=feature_config,
+            flag_identifier=identifier,
             variation=variation
         )
 
@@ -79,7 +79,7 @@ def enqueue(self, target: Target, feature_config: FeatureConfig,
 
     def get_key(self, event: AnalyticsEvent) -> str:
         return '{feature}-{variation}-{value}-{target}'.format(
-            feature=event.feature_config.feature,
+            feature=event.flag_identifier,
             variation=event.variation.identifier,
             value=event.variation.value,
             target=GLOBAL_TARGET,
@@ -122,9 +122,9 @@ def _send_data(self) -> None:
 
                 metric_attributes: List[KeyValue] = [
                     KeyValue(FEATURE_IDENTIFIER_ATTRIBUTE,
-                             event.feature_config.feature),
+                             event.flag_identifier),
                     KeyValue(FEATURE_NAME_ATTRIBUTE,
-                             event.feature_config.feature),
+                             event.flag_identifier),
                     KeyValue(VARIATION_IDENTIFIER_ATTRIBUTE,
                              event.variation.identifier),
                     KeyValue(VARIATION_VALUE_ATTRIBUTE, event.variation.value),
diff --git a/featureflags/client.py b/featureflags/client.py
@@ -6,14 +6,17 @@
 from jwt import decode
 
 from featureflags.analytics import AnalyticsService
+from featureflags.evaluations.evaluator import Evaluator
+from featureflags.lru_cache import LRUCache
+from featureflags.repository import Repository
 
 from .api.client import AuthenticatedClient, Client
 from .api.default.authenticate import AuthenticationRequest
 from .api.default.authenticate import sync as authenticate
 from .config import Config, default_config
-from .evaluations.feature import FeatureConfig
+from .evaluations.feature import FeatureConfig, FeatureConfigKind
 from .evaluations.segment import Segments
-from .evaluations.target import Target
+from .evaluations.auth_target import Target
 from .polling import PollingProcessor
 from .streaming import StreamProcessor
 from .util import log
@@ -39,6 +42,12 @@ def __init__(
             if callable(option):
                 option(self._config)
 
+        if self._config.cache is None:
+            raise Exception("cache cannot be none")
+
+        self._repository = Repository(self._config.cache)
+        self._evaluator = Evaluator(self._repository)
+
         log.debug("CfClient initialized")
         self.run()
 
@@ -53,20 +62,21 @@ def run(self):
             config=self._config,
             environment_id=self._environment_id,
             ready=polling_event,
-            stream_ready=streaming_event
+            stream_ready=streaming_event,
+            repository=self._repository
         )
         self._polling_processor.start()
 
         if self._config.enable_stream:
             self._stream = StreamProcessor(
-                cache=self._config.cache,
+                repository=self._repository,
                 client=self._client,
                 environment_id=self._environment_id,
                 api_key=self._sdk_key,
                 token=self._auth_token,
                 config=self._config,
                 ready=streaming_event,
-                cluster=self._cluster
+                cluster=self._cluster,
             )
             self._stream.start()
 
@@ -105,58 +115,37 @@ def authenticate(self):
         )
         self._client.with_headers({"User-Agent": "PythonSDK/" + VERSION})
 
-    def map_segments_from_cache(self, fc: FeatureConfig) -> None:
-        if self._config.cache:
-            segments = fc.get_segment_identifiers()
-            for identifier in segments:
-                try:
-                    segment = self._config.cache.get(f'segments/{identifier}')
-                    if fc.segments is None:
-                        fc.segments = Segments({})
-                    fc.segments[identifier] = segment
-                except KeyError:
-                    log.warning("segment %s not found in cache", identifier)
-
-    def _variation(self, fn: str, identifier: str, target: Target,
-                   default: Any) -> Any:
-        if self._config.cache:
-            try:
-                fc = self._config.cache.get(f'flags/{identifier}')
-                if fc:
-                    self.map_segments_from_cache(fc)
-                    method = getattr(fc, f'{fn}_variation', None)
-                    if method:
-                        variation = method(target)
-                        if variation is None:
-                            log.debug('No variation found')
-                            return default
-                        self._analytics.enqueue(target, fc, variation)
-                        return getattr(variation, fn)(default)
-                    else:
-                        log.error("Wrong method name %s", fn)
-            except KeyError:
-                log.warning("flag %s not found in cache", identifier)
-        return default
-
     def bool_variation(self, identifier: str, target: Target,
                        default: bool) -> bool:
-        return self._variation('bool', identifier, target, default)
+        variation = self._evaluator.evaluate(identifier, target)
+        self._analytics.enqueue(target, identifier, variation)
+        return variation.bool(default)
 
     def int_variation(self, identifier: str, target: Target,
                       default: int) -> int:
-        return self._variation('int', identifier, target, default)
+        variation = self._evaluator.evaluate(identifier, target)
+        self._analytics.enqueue(target, identifier, variation)
+        return variation.int(default)
 
     def number_variation(self, identifier: str, target: Target,
                          default: float) -> float:
-        return self._variation('number', identifier, target, default)
+        variation = self._evaluator.evaluate(
+            identifier, target)
+        self._analytics.enqueue(target, identifier, variation)
+        return variation.number(default)
 
     def string_variation(self, identifier: str, target: Target,
                          default: str) -> str:
-        return self._variation('string', identifier, target, default)
+        variation = self._evaluator.evaluate(
+            identifier, target)
+        self._analytics.enqueue(target, identifier, variation)
+        return variation.string(default)
 
     def json_variation(self, identifier: str, target: Target,
                        default: Dict[str, Any]) -> Dict[str, Any]:
-        return self._variation('json', identifier, target, default)
+        variation = self._evaluator.evaluate(identifier, target)
+        self._analytics.enqueue(target, identifier, variation)
+        return variation.json(default)
 
     def close(self):
         log.info('closing sdk client')
diff --git a/featureflags/evaluations/evaluator.py b/featureflags/evaluations/evaluator.py
@@ -23,20 +23,23 @@
 from featureflags.util import log
 
 
+EMPTY_VARIATION = Variation(identifier="", value=None)
+
+
 class Evaluator(object):
 
     def __init__(self, provider: QueryInterface):
         self.provider = provider
 
     def _find_variation(self, variations: List[Variation],
-                        identifier: Optional[str]) -> Optional[Variation]:
+                        identifier: Optional[str]) -> Variation:
         if not identifier:
             log.debug("Empty identifier %s or variations %s occurred",
                       identifier, variations)
-            return None
+            return EMPTY_VARIATION
         variation = next(
             (val for val in variations if val.identifier == identifier),
-            None
+            EMPTY_VARIATION
         )
         log.debug("Variation %s found in variations %s",
                   identifier, variations)
@@ -236,7 +239,7 @@ def _evaluate_variation_map(self, var_target_map: List[VariationMap],
         return None
 
     def _evaluate_flag(self, fc: FeatureConfig,
-                       target: Target) -> Optional[Variation]:
+                       target: Target) -> Variation:
         variation: Optional[str] = fc.off_variation
         log.debug("feature %s state is %s", fc.feature, fc.state)
         if fc.state == FeatureState.ON:
@@ -297,11 +300,10 @@ def _check_prerequisite(self, parent: FeatureConfig,
                     return self._check_prerequisite(config, target)
         return True
 
-    def evaluate(self, identifier: str, target: Target,
-                 expected: FeatureConfigKind) -> Optional[Variation]:
+    def evaluate(self, identifier: str, target: Target) -> Variation:
         fc = self.provider.get_flag(identifier)
-        if not fc or fc.kind != expected:
-            return None
+        if not fc:
+            return Variation(identifier="", value=None)
 
         if fc.prerequisites:
             prereq = self._check_prerequisite(fc, target)
diff --git a/featureflags/evaluations/variation.py b/featureflags/evaluations/variation.py
@@ -1,5 +1,6 @@
 import json
 from typing import Any, Dict, List, Type, TypeVar, Union
+from xmlrpc.client import boolean
 
 import attr
 
@@ -11,14 +12,14 @@
 @attr.s(auto_attribs=True)
 class Variation(object):
     identifier: str
-    value: str
+    value: Union[str, None]
     name: Union[Unset, str] = UNSET
     description: Union[Unset, str] = UNSET
     additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
 
     def bool(self, default: bool = False) -> bool:
         if self.value:
-            return self.value == "true"
+            return self.value.lower() == "true"
         return default
 
     def string(self, default: str) -> str:
@@ -94,5 +95,5 @@ def __setitem__(self, key: str, value: Any) -> None:
     def __delitem__(self, key: str) -> None:
         del self.additional_properties[key]
 
-    def __contains__(self, key: str) -> bool:
+    def __contains__(self, key: str) -> boolean:
         return key in self.additional_properties
diff --git a/featureflags/polling.py b/featureflags/polling.py
@@ -2,6 +2,7 @@
 from threading import Event, Thread
 
 from featureflags.api.client import AuthenticatedClient
+from featureflags.repository import DataProviderInterface
 
 from .api.default.get_all_segments import sync as retrieve_segments
 from .api.default.get_feature_config import sync as retrieve_flags
@@ -13,7 +14,7 @@ class PollingProcessor(Thread):
 
     def __init__(self, client: AuthenticatedClient, config: Config,
                  environment_id: str, ready: Event,
-                 stream_ready: Event) -> None:
+                 stream_ready: Event, repository: DataProviderInterface) -> None:
         Thread.__init__(self)
         self.daemon = True
         self.__environment_id = environment_id
@@ -22,6 +23,7 @@ def __init__(self, client: AuthenticatedClient, config: Config,
         self.__running = False
         self.__ready = ready
         self.__stream_ready = stream_ready
+        self.__repository = repository
 
     def run(self):
         if not self.__running:
@@ -66,8 +68,8 @@ def __retrieve_flags(self):
         )
         log.debug("Feature flags loaded")
         for flag in flags:
-            log.debug("Setting the cache value %s", flag.feature)
-            self.__config.cache.set(f"flags/{flag.feature}", flag)
+            log.debug("Put flag %s into repository", flag.feature)
+            self.__repository.set_flag(flag)
 
     def __retrieve_segments(self):
         log.debug("Loading target segments")
@@ -76,5 +78,5 @@ def __retrieve_segments(self):
         )
         log.debug("Target segments loaded")
         for segment in segments:
-            log.debug("Setting the cache segment value %s", segment.identifier)
-            self.__config.cache.set(f"segments/{segment.identifier}", segment)
+            log.debug("Put %s segment into repository", segment.identifier)
+            self.__repository.set_segment(segment)
diff --git a/featureflags/repository.py b/featureflags/repository.py
@@ -1,5 +1,6 @@
 import abc
 from typing import List, Optional
+from featureflags.evaluations.constants import SEGMENT_MATCH_OPERATOR
 
 from featureflags.evaluations.feature import FeatureConfig
 from featureflags.evaluations.segment import Segment
@@ -58,6 +59,16 @@ def set_segment(self, group: Segment) -> None:
         """Put Target group to the repository"""
         raise NotImplementedError
 
+    @abc.abstractmethod
+    def remove_flag(self, identifier: str) -> None:
+        """Remove Flag from the repository"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def remove_segment(self, identifier: str) -> None:
+        """Remove Target group from the repository"""
+        raise NotImplementedError
+
     @abc.abstractmethod
     def close(self) -> None:
         """Put Target group to the repository"""
@@ -140,8 +151,45 @@ def set_segment(self, segment: Segment) -> None:
             self.cache.set(segment_key, segment)
             log.debug("Segment %s successfully cached", segment.identifier)
 
-    def find_flags_by_segment(self, identifier: str) -> List[str]:
-        return []
+    def find_flags_by_segment(self, segment: str) -> List[str]:
+        result = []
+        keys = self.cache.keys()
+        if self.store:
+            keys = self.store.keys()
+        for key in keys:
+            flag = self.get_flag(key, cacheable=False)
+            if not flag:
+                continue
+            for serving_rule in flag.rules:
+                for clause in serving_rule.clauses:
+                    if clause.op == SEGMENT_MATCH_OPERATOR and not next(
+                        (val for val in clause.values if val == segment),
+                        None
+                    ):
+                        log.debug("Flag %s evaluated in segments",
+                                  flag.feature)
+                        result.append(flag.feature)
+        return result
+
+    def remove_flag(self, identifier: str) -> None:
+        """Remove Flag from the repository"""
+        flag_key = format_flag_key(identifier)
+        if self.store:
+            self.store.remove([flag_key])
+            log.debug("Flag %s successfully deleted from store", identifier)
+
+        self.cache.remove([flag_key])
+        log.debug("Flag %s successfully deleted from cache", identifier)
+
+    def remove_segment(self, identifier: str) -> None:
+        """Remove Target group from the repository"""
+        segment_key = format_segment_key(identifier)
+        if self.store:
+            self.store.remove([segment_key])
+            log.debug("Segment %s successfully deleted from store", identifier)
+
+        self.cache.remove([segment_key])
+        log.debug("Segment %s successfully deleted from cache", identifier)
 
     def close(self) -> None:
         if self.store:
diff --git a/featureflags/streaming.py b/featureflags/streaming.py
diff --git a/tests/integration/test_evaluator.py b/tests/integration/test_evaluator.py
diff --git a/tests/unit/test_evaluator.py b/tests/unit/test_evaluator.py
