feat: Add async big segment store manager and async Redis adapter

ldclient/async_config.py

@@ -0,0 +1,65 @@
+"""
+Configuration types for the async SDK client. Currently this holds
+:class:`AsyncBigSegmentsConfig`; the full :class:`AsyncConfig` lands in a later slice.
+
+.. caution::
+    This feature is experimental and should NOT be considered ready for production
+    use. It may change or be removed without notice and is not subject to backwards
+    compatibility guarantees.
+"""
+
+from typing import Optional
+
+from ldclient.interfaces import AsyncBigSegmentStore
+
+
+class AsyncBigSegmentsConfig:
+    """Configuration options related to Big Segments for the async SDK client.
+
+    Big Segments are a specific type of segments. For more information, read the LaunchDarkly
+    documentation: https://docs.launchdarkly.com/home/users/big-segments
+
+    If your application uses Big Segments, you will need to create an ``AsyncBigSegmentsConfig``
+    that at a minimum specifies what database integration to use, and then pass the
+    ``AsyncBigSegmentsConfig`` object as the ``big_segments`` parameter when creating an
+    :class:`AsyncConfig`.
+    """
+
+    def __init__(self, store: Optional[AsyncBigSegmentStore] = None, context_cache_size: int = 1000, context_cache_time: float = 5, status_poll_interval: float = 5, stale_after: float = 120):
+        """
+        :param store: the implementation of :class:`ldclient.interfaces.AsyncBigSegmentStore` that
+            will be used to query the Big Segments database
+        :param context_cache_size: the maximum number of contexts whose Big Segment state will be cached
+            by the SDK at any given time
+        :param context_cache_time: the maximum length of time (in seconds) that the Big Segment state
+            for a context will be cached by the SDK
+        :param status_poll_interval: the interval (in seconds) at which the SDK will poll the Big
+            Segment store to make sure it is available and to determine how long ago it was updated
+        :param stale_after: the maximum length of time between updates of the Big Segments data
+            before the data is considered out of date
+        """
+        self.__store = store
+        self.__context_cache_size = context_cache_size
+        self.__context_cache_time = context_cache_time
+        self.__status_poll_interval = status_poll_interval
+        self.__stale_after = stale_after
+
+    @property
+    def store(self) -> Optional[AsyncBigSegmentStore]:
+        return self.__store
+
+    @property
+    def context_cache_size(self) -> int:
+        return self.__context_cache_size
+
+    @property
+    def context_cache_time(self) -> float:
+        return self.__context_cache_time
+
+    @property
+    def status_poll_interval(self) -> float:
+        return self.__status_poll_interval
+
+    @property
+    def stale_after(self) -> float:
+        return self.__stale_after

ldclient/impl/async_big_segments.py

@@ -0,0 +1,98 @@
+from typing import Optional, Tuple
+
+from expiringdict import ExpiringDict
+
+from ldclient.async_config import AsyncBigSegmentsConfig
+from ldclient.evaluation import BigSegmentsStatus
+from ldclient.impl.aio.concurrency import AsyncRepeatingTask
+from ldclient.impl.big_segments_common import (
+    EMPTY_MEMBERSHIP,
+    BigSegmentStoreStatusProviderImpl,
+    _hash_for_user_key,
+    is_stale
+)
+from ldclient.impl.util import log
+from ldclient.interfaces import (
+    BigSegmentStoreStatus,
+    BigSegmentStoreStatusProvider
+)
+
+
+class AsyncBigSegmentStoreManager:
+    """
+    Internal component that decorates the Big Segment store with caching behavior, and also polls the
+    store to track its status. The constructor starts the polling task.
+    """
+
+    # Because the constructor starts the polling task, it must run within a running event loop.
+    def __init__(self, config: AsyncBigSegmentsConfig):
+        self.__store = config.store
+
+        self.__stale_after_millis = config.stale_after * 1000
+        self.__status_provider = BigSegmentStoreStatusProviderImpl(self.get_status)
+        self.__last_status = None  # type: Optional[BigSegmentStoreStatus]
+        self.__poll_task = None  # type: Optional[AsyncRepeatingTask]
+
+        if self.__store:
+            self.__cache = ExpiringDict(max_len=config.context_cache_size, max_age_seconds=config.context_cache_time)
+            self.__poll_task = AsyncRepeatingTask("ldclient.bigsegment.status-poll", config.status_poll_interval, 0, self.poll_store_and_update_status)
+            self.__poll_task.start()
+
+    async def stop(self):
+        if self.__poll_task:
+            self.__poll_task.stop()
+        if self.__store:
+            await self.__store.stop()
+
+    @property
+    def status_provider(self) -> BigSegmentStoreStatusProvider:
+        return self.__status_provider
+
+    async def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str]:
+        if not self.__store:
+            return None, BigSegmentsStatus.NOT_CONFIGURED
+        membership = self.__cache.get(user_key)
+        if membership is None:
+            user_hash = _hash_for_user_key(user_key)
+            try:
+                membership = await self.__store.get_membership(user_hash)  # type: ignore[misc]
+                if membership is None:
+                    membership = EMPTY_MEMBERSHIP
+                self.__cache[user_key] = membership
+            except Exception as e:
+                log.exception("Big Segment store membership query returned error: %s" % e)
+                return None, BigSegmentsStatus.STORE_ERROR
+        # First-call fallback: if the polling task hasn't run yet, poll inline now
+        status = self.__last_status
+        if status is None:
+            status = await self.poll_store_and_update_status()
+        if not status.available:
+            return membership, BigSegmentsStatus.STORE_ERROR
+        return membership, BigSegmentsStatus.STALE if status.stale else BigSegmentsStatus.HEALTHY
+
+    def get_status(self) -> BigSegmentStoreStatus:
+        """Return the most recently polled status.
+
+        When no status has been cached yet, the sync variant polls the store
+        inline; the async variant (whose status getter cannot await) reports
+        the store as unavailable until the polling task has run.
+        """
+        status = self.__last_status
+        if status is None:
+            return BigSegmentStoreStatus(False, False)
+        return status if status else self.poll_store_and_update_status()  # type: ignore[return-value]
+
+    async def poll_store_and_update_status(self) -> BigSegmentStoreStatus:
+        new_status = BigSegmentStoreStatus(False, False)  # default to "unavailable" if we don't get a new status below
+        if self.__store:
+            try:
+                metadata = await self.__store.get_metadata()  # type: ignore[misc]
+                new_status = BigSegmentStoreStatus(True, (metadata is None) or self.is_stale(metadata.last_up_to_date))
+            except Exception as e:
+                log.exception("Big Segment store status query returned error: %s" % e)
+        self.__last_status = new_status
+        self.__status_provider._update_status(new_status)
+        return new_status
+
+    def is_stale(self, timestamp) -> bool:
+        return is_stale(timestamp, self.__stale_after_millis)

ldclient/impl/big_segments.py

@@ -1,13 +1,15 @@
-import base64
-import time
-from hashlib import sha256
-from typing import Callable, Optional, Tuple
+from typing import Optional, Tuple
 
 from expiringdict import ExpiringDict
 
 from ldclient.config import BigSegmentsConfig
 from ldclient.evaluation import BigSegmentsStatus
-from ldclient.impl.listeners import Listeners
+from ldclient.impl.big_segments_common import (
+    EMPTY_MEMBERSHIP,
+    BigSegmentStoreStatusProviderImpl,
+    _hash_for_user_key,
+    is_stale
+)
 from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.util import log
 from ldclient.interfaces import (
@@ -16,44 +18,7 @@
 )
 
 
-class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider):
-    """
-    Default implementation of the BigSegmentStoreStatusProvider interface.
-
-    The real implementation of getting the status is in BigSegmentStoreManager - we pass in a lambda that
-    allows us to get the current status from that class. So this class provides a facade for that, and
-    also adds the listener mechanism.
-    """
-
-    def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]):
-        self.__status_getter = status_getter
-        self.__status_listeners = Listeners()
-        self.__last_status = None  # type: Optional[BigSegmentStoreStatus]
-
-    @property
-    def status(self) -> BigSegmentStoreStatus:
-        return self.__status_getter()
-
-    def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None:
-        self.__status_listeners.add(listener)
-
-    def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None:
-        self.__status_listeners.remove(listener)
-
-    def _update_status(self, new_status: BigSegmentStoreStatus):
-        last = self.__last_status
-        if last is None:
-            self.__last_status = new_status
-        elif new_status.available != last.available or new_status.stale != last.stale:
-            self.__last_status = new_status
-            self.__status_listeners.notify(new_status)
-
-
 class BigSegmentStoreManager:
-    # use EMPTY_MEMBERSHIP as a singleton whenever a membership query returns None; it's safe to reuse it
-    # because we will never modify the membership properties after they're queried
-    EMPTY_MEMBERSHIP = {}  # type: dict
-
     """
     Internal component that decorates the Big Segment store with caching behavior, and also polls the
     store to track its status.
@@ -91,7 +56,7 @@ def get_user_membership(self, user_key: str) -> Tuple[Optional[dict], str]:
             try:
                 membership = self.__store.get_membership(user_hash)
                 if membership is None:
-                    membership = self.EMPTY_MEMBERSHIP
+                    membership = EMPTY_MEMBERSHIP
                 self.__cache[user_key] = membership
             except Exception as e:
                 log.exception("Big Segment store membership query returned error: %s" % e)
@@ -120,8 +85,4 @@ def poll_store_and_update_status(self) -> BigSegmentStoreStatus:
         return new_status
 
     def is_stale(self, timestamp) -> bool:
-        return (timestamp is None) or ((int(time.time() * 1000) - timestamp) >= self.__stale_after_millis)
-
-
-def _hash_for_user_key(user_key: str) -> str:
-    return base64.b64encode(sha256(user_key.encode('utf-8')).digest()).decode('utf-8')
+        return is_stale(timestamp, self.__stale_after_millis)

ldclient/impl/big_segments_common.py

@@ -0,0 +1,63 @@
+"""
+Shared, I/O-free big-segments status provider used by both the sync
+:mod:`ldclient.impl.big_segments` and the async
+:mod:`ldclient.impl.async_big_segments`. It holds the last known status and
+notifies listeners; nothing here touches the store or network, so it is
+identical across the two managers.
+"""
+
+import base64
+import time
+from hashlib import sha256
+from typing import Callable, Optional
+
+from ldclient.impl.listeners import Listeners
+from ldclient.interfaces import (
+    BigSegmentStoreStatus,
+    BigSegmentStoreStatusProvider
+)
+
+# use EMPTY_MEMBERSHIP as a singleton whenever a membership query returns None; it's safe to reuse it
+# because we will never modify the membership properties after they're queried
+EMPTY_MEMBERSHIP = {}  # type: dict
+
+
+def _hash_for_user_key(user_key: str) -> str:
+    return base64.b64encode(sha256(user_key.encode('utf-8')).digest()).decode('utf-8')
+
+
+def is_stale(timestamp: int, stale_after_millis) -> bool:
+    return (timestamp is None) or ((int(time.time() * 1000) - timestamp) >= stale_after_millis)
+
+
+class BigSegmentStoreStatusProviderImpl(BigSegmentStoreStatusProvider):
+    """
+    Default implementation of the BigSegmentStoreStatusProvider interface.
+
+    The real implementation of getting the status is in the big segment store manager - we pass in a lambda that
+    allows us to get the current status from that class. So this class provides a facade for that, and
+    also adds the listener mechanism.
+    """
+
+    def __init__(self, status_getter: Callable[[], BigSegmentStoreStatus]):
+        self.__status_getter = status_getter
+        self.__status_listeners = Listeners()
+        self.__last_status = None  # type: Optional[BigSegmentStoreStatus]
+
+    @property
+    def status(self) -> BigSegmentStoreStatus:
+        return self.__status_getter()
+
+    def add_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None:
+        self.__status_listeners.add(listener)
+
+    def remove_listener(self, listener: Callable[[BigSegmentStoreStatus], None]) -> None:
+        self.__status_listeners.remove(listener)
+
+    def _update_status(self, new_status: BigSegmentStoreStatus):
+        last = self.__last_status
+        if last is None:
+            self.__last_status = new_status
+        elif new_status.available != last.available or new_status.stale != last.stale:
+            self.__last_status = new_status
+            self.__status_listeners.notify(new_status)

ldclient/impl/integrations/redis/async_redis_big_segment_store.py

@@ -0,0 +1,41 @@
+from typing import Any, Dict, Optional
+
+from ldclient.impl.util import log, redact_password
+from ldclient.interfaces import AsyncBigSegmentStore, BigSegmentStoreMetadata
+
+have_async_redis = False
+try:
+    import redis.asyncio as redis_client
+
+    have_async_redis = True
+except ImportError:
+    pass
+
+
+class _AsyncRedisBigSegmentStore(AsyncBigSegmentStore):
+    KEY_LAST_UP_TO_DATE = ':big_segments_synchronized_on'
+    KEY_USER_INCLUDE = ':big_segment_include:'
+    KEY_USER_EXCLUDE = ':big_segment_exclude:'
+
+    def __init__(self, url: str, prefix: Optional[str], redis_opts: Dict[str, Any]):
+        if not have_async_redis:
+            raise NotImplementedError("Cannot use async Redis Big Segment store because redis package is not installed")
+        self._prefix = prefix or 'launchdarkly'
+        self._client = redis_client.from_url(url, **redis_opts)
+        log.info("Started AsyncRedisBigSegmentStore connected to URL: " + redact_password(url) + " using prefix: " + self._prefix)
+
+    async def get_metadata(self) -> BigSegmentStoreMetadata:
+        value = await self._client.get(self._prefix + self.KEY_LAST_UP_TO_DATE)
+        return BigSegmentStoreMetadata(int(value) if value else None)
+
+    async def get_membership(self, user_hash: str) -> Optional[dict]:
+        included = await self._client.smembers(self._prefix + self.KEY_USER_INCLUDE + user_hash)
+        excluded = await self._client.smembers(self._prefix + self.KEY_USER_EXCLUDE + user_hash)
+        if not included and not excluded:
+            return None
+        ret = {ref.decode(): False for ref in excluded}
+        ret.update({ref.decode(): True for ref in included})
+        return ret
+
+    async def stop(self):
+        await self._client.aclose()

ldclient/integrations/__init__.py

@@ -237,6 +237,40 @@ def new_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: str = '
 
         return _RedisBigSegmentStore(url, prefix, redis_opts)
 
+    @staticmethod
+    def async_big_segment_store(url: str = 'redis://localhost:6379/0', prefix: Optional[str] = None, redis_opts: Dict[str, Any] = {}):
+        """
+        Creates an async Redis-backed Big Segment store implementing :class:`~ldclient.interfaces.AsyncBigSegmentStore`.
+
+        .. caution::
+            This feature is experimental and should NOT be considered ready for production
+            use. It may change or be removed without notice and is not subject to backwards
+            compatibility guarantees. Pin to a specific minor version and review the changelog
+            before upgrading.
+
+        Big Segments are a specific type of user segments. For more information, read the LaunchDarkly
+        documentation: https://docs.launchdarkly.com/home/users/big-segments
+
+        To use this method, you must first install the ``redis`` package (version >=5.0.1). Then, put
+        the object returned by this method into the ``store`` property of your Big Segments
+        configuration when constructing an ``AsyncLDClient``.
+        ::
+
+          from ldclient.config import Config, BigSegmentsConfig
+          from ldclient.integrations import Redis
+          store = Redis.async_big_segment_store()
+          config = Config(big_segments=BigSegmentsConfig(store=store))
+
+        :param url: the URL of the Redis host; defaults to ``DEFAULT_URL``
+        :param prefix: a namespace prefix to be prepended to all Redis keys; defaults to
+          ``DEFAULT_PREFIX``
+        :param redis_opts: extra options forwarded to ``redis.asyncio.from_url``
+        """
+        from ldclient.impl.integrations.redis.async_redis_big_segment_store import (
+            _AsyncRedisBigSegmentStore
+        )
+        return _AsyncRedisBigSegmentStore(url, prefix, redis_opts)
+
 
 class Files:
     """Provides factory methods for integrations with filesystem data."""