chore: refactor connector api so that an async variant can be created

Author: sighphyre

UnleashClient/__init__.py

@@ -17,7 +17,7 @@
 
 from UnleashClient.api.sync_api import register_client
 from UnleashClient.connectors import (
-    BaseConnector,
+    BaseSyncConnector,
     BootstrapConnector,
     OfflineConnector,
     PollingConnector,
@@ -175,7 +175,7 @@ def __init__(
                 cache=self.cache,
             ).start()
 
-        self.connector: BaseConnector = None
+        self.connector: BaseSyncConnector = None
 
         self._evaluator = Evaluator(
             engine=self.engine,

UnleashClient/connectors/__init__.py

@@ -1,11 +1,11 @@
-from .base_connector import BaseConnector
+from .base_sync_connector import BaseSyncConnector
 from .bootstrap_connector import BootstrapConnector
 from .offline_connector import OfflineConnector
 from .polling_connector import PollingConnector
 from .streaming_connector import StreamingConnector
 
 __all__ = [
-    "BaseConnector",
+    "BaseSyncConnector",
     "BootstrapConnector",
     "OfflineConnector",
     "PollingConnector",

UnleashClient/connectors/base_connector.py

@@ -0,0 +1,29 @@
+from abc import ABC, abstractmethod
+from typing import Callable, Optional
+from yggdrasil_engine.engine import UnleashEngine
+from UnleashClient.cache import BaseCache
+
+
+class BaseSyncConnector(ABC):
+    def __init__(
+        self,
+        engine: UnleashEngine,
+        cache: BaseCache,
+        ready_callback: Optional[Callable] = None,
+    ):
+        """
+        :param engine: Feature evaluation engine instance (UnleashEngine).
+        :param cache: Should be the cache class variable from UnleashClient
+        :param ready_callback: Optional function to call when features are successfully loaded.
+        """
+        self.engine = engine
+        self.cache = cache
+        self.ready_callback = ready_callback
+
+    @abstractmethod
+    def start(self):
+        pass
+
+    @abstractmethod
+    def stop(self):
+        pass

UnleashClient/connectors/base_sync_connector.py

@@ -1,11 +1,12 @@
 from yggdrasil_engine.engine import UnleashEngine
 
 from UnleashClient.cache import BaseCache
+from UnleashClient.connectors.hydration import hydrate_engine
 
-from .base_connector import BaseConnector
+from .base_sync_connector import BaseSyncConnector
 
 
-class BootstrapConnector(BaseConnector):
+class BootstrapConnector(BaseSyncConnector):
     def __init__(
         self,
         engine: UnleashEngine,
@@ -16,7 +17,7 @@ def __init__(
         self.job = None
 
     def start(self):
-        self.load_features()
+        hydrate_engine(self.cache, self.engine, None)
 
     def stop(self):
         pass

UnleashClient/connectors/bootstrap_connector.py

@@ -0,0 +1,32 @@
+from typing import Callable, Optional
+
+from yggdrasil_engine.engine import UnleashEngine
+
+from UnleashClient.cache import BaseCache
+from UnleashClient.constants import FEATURES_URL
+from UnleashClient.utils import LOGGER
+
+
+def hydrate_engine(
+    cache: BaseCache, engine: UnleashEngine, ready_callback: Optional[Callable] = None
+):
+    feature_provisioning = cache.get(FEATURES_URL)
+    if not feature_provisioning:
+        LOGGER.warning(
+            "Unleash client does not have cached features. "
+            "Please make sure client can communicate with Unleash server!"
+        )
+        return
+
+    try:
+        warnings = engine.take_state(feature_provisioning)
+        if ready_callback:
+            ready_callback()
+        if warnings:
+            LOGGER.warning(
+                "Some features were not able to be parsed correctly, they may not evaluate as expected"
+            )
+            LOGGER.warning(warnings)
+    except Exception as e:
+        LOGGER.error(f"Error loading features: {e}")
+        LOGGER.debug(f"Full feature response body from server: {feature_provisioning}")

UnleashClient/connectors/hydration.py

@@ -5,11 +5,12 @@
 from yggdrasil_engine.engine import UnleashEngine
 
 from UnleashClient.cache import BaseCache
+from UnleashClient.connectors.hydration import hydrate_engine
 
-from .base_connector import BaseConnector
+from .base_sync_connector import BaseSyncConnector
 
 
-class OfflineConnector(BaseConnector):
+class OfflineConnector(BaseSyncConnector):
     def __init__(
         self,
         engine: UnleashEngine,
@@ -29,11 +30,14 @@ def __init__(
         self.refresh_jitter = refresh_jitter
         self.job = None
 
+    def hydrate(self):
+        hydrate_engine(self.cache, self.engine, self.ready_callback)
+
     def start(self):
-        self.load_features()
+        self.hydrate()
 
         self.job = self.scheduler.add_job(
-            self.load_features,
+            self.hydrate,
             trigger=IntervalTrigger(
                 seconds=self.refresh_interval, jitter=self.refresh_jitter
             ),

UnleashClient/connectors/offline_connector.py

@@ -7,14 +7,15 @@
 
 from UnleashClient.api.sync_api import get_feature_toggles
 from UnleashClient.cache import BaseCache
+from UnleashClient.connectors.hydration import hydrate_engine
 from UnleashClient.constants import ETAG, FEATURES_URL
 from UnleashClient.events import UnleashEventType, UnleashFetchedEvent
 from UnleashClient.utils import LOGGER
 
-from .base_connector import BaseConnector
+from .base_sync_connector import BaseSyncConnector
 
 
-class PollingConnector(BaseConnector):
+class PollingConnector(BaseSyncConnector):
     def __init__(
         self,
         engine: UnleashEngine,
@@ -78,7 +79,7 @@ def _fetch_and_load(self):
         if etag:
             self.cache.set(ETAG, etag)
 
-        self.load_features()
+        hydrate_engine(self.cache, self.engine, self.ready_callback)
 
         if state:
             if self.event_callback:

UnleashClient/connectors/polling_connector.py

@@ -6,12 +6,13 @@
 from yggdrasil_engine.engine import UnleashEngine
 
 from UnleashClient.cache import BaseCache
-from UnleashClient.connectors.base_connector import BaseConnector
+from UnleashClient.connectors.base_sync_connector import BaseSyncConnector
+from UnleashClient.connectors.hydration import hydrate_engine
 from UnleashClient.constants import APPLICATION_HEADERS, FEATURES_URL, STREAMING_URL
 from UnleashClient.utils import LOGGER
 
 
-class StreamingConnector(BaseConnector):
+class StreamingConnector(BaseSyncConnector):
     def __init__(
         self,
         engine: UnleashEngine,
@@ -109,14 +110,14 @@ def _run(self):
                                 LOGGER.debug("Ready callback failed", exc_info=True)
                     except Exception:
                         LOGGER.error("Error applying streaming state", exc_info=True)
-                        self.load_features()
+                        hydrate_engine(self.cache, self.engine, self.ready_callback)
                 else:
                     LOGGER.debug("Ignoring SSE event type: %s", event.event)
 
             LOGGER.debug("SSE stream ended")
         except Exception as exc:
             LOGGER.warning("Streaming connection failed: %s", exc)
-            self.load_features()
+            hydrate_engine(self.cache, self.engine, self.ready_callback)
         finally:
             try:
                 if self._client is not None:

UnleashClient/connectors/streaming_connector.py

@@ -3,6 +3,7 @@
 from apscheduler.schedulers.background import BackgroundScheduler
 from yggdrasil_engine.engine import UnleashEngine
 
+from UnleashClient.connectors.hydration import hydrate_engine
 from tests.utilities.mocks.mock_features import MOCK_FEATURE_RESPONSE
 from UnleashClient.connectors import OfflineConnector
 from UnleashClient.constants import FEATURES_URL
@@ -21,7 +22,7 @@ def test_offline_connector_load_features(cache_empty):
         scheduler=scheduler,
     )
 
-    connector.load_features()
+    hydrate_engine(connector.cache, connector.engine, None)
     assert engine.is_enabled("testFlag", {})