feat: add odp manager

optimizely/exceptions.py

@@ -64,3 +64,21 @@ class UnsupportedDatafileVersionException(Exception):
     """ Raised when provided version in datafile is not supported. """
 
     pass
+
+
+class OdpNotEnabled(Exception):
+    """ Raised when Optimizely Data Platform (ODP) is not enabled. """
+
+    pass
+
+
+class OdpNotIntegrated(Exception):
+    """ Raised when Optimizely Data Platform (ODP) is not integrated. """
+
+    pass
+
+
+class OdpInvalidData(Exception):
+    """ Raised when passing invalid ODP data. """
+
+    pass

optimizely/helpers/enums.py

@@ -120,11 +120,12 @@ class Errors:
     NONE_VARIABLE_KEY_PARAMETER: Final = '"None" is an invalid value for variable key.'
     UNSUPPORTED_DATAFILE_VERSION: Final = (
         'This version of the Python SDK does not support the given datafile version: "{}".')
-    INVALID_SEGMENT_IDENTIFIER: Final = 'Audience segments fetch failed (invalid identifier).'
-    FETCH_SEGMENTS_FAILED: Final = 'Audience segments fetch failed ({}).'
-    ODP_EVENT_FAILED: Final = 'ODP event send failed ({}).'
-    ODP_NOT_ENABLED: Final = 'ODP is not enabled.'
-    ODP_NOT_INTEGRATED: Final = 'ODP is not integrated.'
+    INVALID_SEGMENT_IDENTIFIER = 'Audience segments fetch failed (invalid identifier).'
+    FETCH_SEGMENTS_FAILED = 'Audience segments fetch failed ({}).'
+    ODP_EVENT_FAILED = 'ODP event send failed ({}).'
+    ODP_NOT_INTEGRATED = 'ODP is not integrated.'
+    ODP_NOT_ENABLED = 'ODP is not enabled.'
+    ODP_INVALID_DATA = 'ODP data is not valid.'
 
 
 class ForcedDecisionLogs:
@@ -214,3 +215,9 @@ class OdpEventManagerConfig:
     DEFAULT_BATCH_SIZE: Final = 10
     DEFAULT_FLUSH_INTERVAL: Final = 1
     DEFAULT_RETRY_COUNT: Final = 3
+
+
+class OdpManagerConfig:
+    """ODP Manager configs."""
+    KEY_FOR_USER_ID = 'fs_user_id'
+    EVENT_TYPE = 'fullstack'

optimizely/odp/odp_event_manager.py

@@ -12,17 +12,18 @@
 # limitations under the License.
 
 from __future__ import annotations
+
+import time
 from enum import Enum
+from queue import Empty, Queue, Full
 from threading import Thread
 from typing import Optional
-import time
-from queue import Empty, Queue, Full
 
 from optimizely import logger as _logging
-from .odp_event import OdpEvent, OdpDataDict
+from optimizely.helpers.enums import OdpEventManagerConfig, Errors, OdpManagerConfig
 from .odp_config import OdpConfig, OdpConfigState
+from .odp_event import OdpEvent, OdpDataDict
 from .zaius_rest_api_manager import ZaiusRestApiManager
-from optimizely.helpers.enums import OdpEventManagerConfig, Errors
 
 
 class Signal(Enum):
@@ -41,10 +42,10 @@ class OdpEventManager:
     """
 
     def __init__(
-        self,
-        odp_config: OdpConfig,
-        logger: Optional[_logging.Logger] = None,
-        api_manager: Optional[ZaiusRestApiManager] = None
+            self,
+            odp_config: OdpConfig,
+            logger: Optional[_logging.Logger] = None,
+            api_manager: Optional[ZaiusRestApiManager] = None
     ):
         """OdpEventManager init method to configure event batching.
 
@@ -236,3 +237,7 @@ def dispatch(self, event: OdpEvent) -> None:
             self.event_queue.put_nowait(event)
         except Full:
             self.logger.warning(Errors.ODP_EVENT_FAILED.format("Queue is full"))
+
+    def identify_user(self, user_id: str) -> None:
+        self.send_event(OdpManagerConfig.EVENT_TYPE, 'identified',
+                        {OdpManagerConfig.KEY_FOR_USER_ID: user_id}, {})

optimizely/odp/odp_manager.py

@@ -0,0 +1,124 @@
+# Copyright 2022, Optimizely
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from typing import Optional, Any
+
+from optimizely import exceptions as optimizely_exception
+from optimizely import logger as optimizely_logger
+from optimizely.helpers.enums import Errors, OdpManagerConfig
+from optimizely.helpers.validator import are_odp_data_types_valid
+from optimizely.odp.lru_cache import LRUCache
+from optimizely.odp.odp_config import OdpConfig
+from optimizely.odp.odp_event_manager import OdpEventManager
+from optimizely.odp.odp_segment_manager import OdpSegmentManager
+from optimizely.odp.zaius_graphql_api_manager import ZaiusGraphQLApiManager
+
+
+class OdpManager:
+    """Orchestrates segment manager, event manager and odp config."""
+
+    def __init__(self, disable: bool,
+                 cache_size: int,
+                 cache_timeout_in_sec: int,
+                 segment_manager: Optional[OdpSegmentManager] = None,
+                 event_manager: Optional[OdpEventManager] = None,
+                 logger: Optional[optimizely_logger.Logger] = None) -> None:
+
+        self.enabled = not disable
+        self.cache_size = cache_size
+        self.cache_timeout_in_sec = cache_timeout_in_sec
+        self.odp_config = OdpConfig()
+        self.logger = logger or optimizely_logger.NoOpLogger()
+
+        self.segment_manager = None
+        self.event_manager = None
+
+        if self.enabled:
+            if segment_manager:
+                segment_manager.odp_config = self.odp_config
+                self.segment_manager = segment_manager
+            else:
+                self.segment_manager = OdpSegmentManager(self.odp_config,
+                                                         LRUCache(cache_size, cache_timeout_in_sec),
+                                                         ZaiusGraphQLApiManager(), logger)
+            if event_manager:
+                event_manager.odp_config = self.odp_config
+                self.event_manager = event_manager
+            else:
+                self.event_manager = OdpEventManager(self.odp_config)
+
+    def fetch_qualified_segments(self, user_id: str, options: list[str]) -> None:
+        if not self.enabled:
+            self.logger.debug(Errors.ODP_NOT_ENABLED)
+            return None
+
+        user_key = OdpManagerConfig.KEY_FOR_USER_ID
+        user_value = user_id
+
+        if self.segment_manager:
+            self.segment_manager.fetch_qualified_segments(user_key, user_value, options)
+
+    def identify_user(self, user_id: str) -> None:
+        if not self.enabled:
+            self.logger.debug('ODP identify event is not dispatched (ODP disabled).')
+
+        if self.event_manager:
+            self.event_manager.identify_user(user_id)
+
+    def send_event(self, type: str, action: str, identifiers: dict[str, str], data: dict[str, Any]) -> None:
+        """
+        Send an event to the ODP server.
+
+        Args:
+            type: The event type.
+            action: The event action name.
+            identifiers: A dictionary for identifiers.
+            data: A dictionary for associated data. The default event data will be added to this data
+            before sending to the ODP server.
+
+        Raises custom exception if error is detected.
+        """
+        if not self.enabled:
+            raise optimizely_exception.OdpNotEnabled(Errors.ODP_NOT_ENABLED)
+
+        if not are_odp_data_types_valid(data):
+            raise optimizely_exception.OdpInvalidData(Errors.ODP_INVALID_DATA)
+
+        if self.event_manager:
+            self.event_manager.send_event(type, action, identifiers, data)
+
+    def update_odp_config(self, api_key: Optional[str], api_host: Optional[str],
+                          segments_to_check: list[str]) -> None:
+        if not self.enabled:
+            return None
+
+        # flush old events using old odp publicKey (if exists) before updating odp key.
+        if self.event_manager:
+            self.event_manager.flush()
+            # wait until done flushing to avoid flushing in the middle of the batch
+            self.event_manager.event_queue.join()
+
+        config_changed = self.odp_config.update(api_key, api_host, segments_to_check)
+        if not config_changed:
+            return None
+
+        # reset segments cache when odp integration or segmentsToCheck are changed
+        if self.segment_manager:
+            self.segment_manager.reset()
+
+        # flush events with the new integration key if events still remain in the queue
+        # (when we get the first datafile ready)
+        if self.event_manager:
+            self.event_manager.flush()

optimizely/odp/odp_segment_manager.py

@@ -64,7 +64,7 @@ def fetch_qualified_segments(self, user_key: str, user_value: str, options: list
         reset_cache = OptimizelyOdpOption.RESET_CACHE in options
 
         if reset_cache:
-            self._reset()
+            self.reset()
 
         if not ignore_cache and not reset_cache:
             segments = self.segments_cache.lookup(cache_key)
@@ -83,7 +83,7 @@ def fetch_qualified_segments(self, user_key: str, user_value: str, options: list
 
         return segments
 
-    def _reset(self) -> None:
+    def reset(self) -> None:
         self.segments_cache.reset()
 
     def make_cache_key(self, user_key: str, user_value: str) -> str: