feat: Add AsyncLDClient with FDv1 data system and public API

ldclient/__init__.py

@@ -88,4 +88,13 @@ def _reset_client():
 __BASE_TYPES__ = (str, float, int, bool)
 
 
+def __getattr__(name):
+    # AsyncLDClient loads lazily so importing ldclient does not require aiohttp
+    # unless you use the async client.
+    if name == 'AsyncLDClient':
+        from ldclient.async_client import AsyncLDClient
+        return AsyncLDClient
+    raise AttributeError("module 'ldclient' has no attribute %r" % name)
+
+
 __all__ = ['Config', 'Context', 'ContextBuilder', 'ContextMultiBuilder', 'LDClient', 'Result', 'client', 'context', 'evaluation', 'integrations', 'interfaces', 'migrations']

ldclient/client.py

@@ -238,7 +238,7 @@ def postfork(self, start_wait: float = 5):
 
     def __start_up(self, start_wait: float):
         environment_metadata = get_environment_metadata(self._config, "python-server-sdk")
-        plugin_hooks = get_plugin_hooks(self._config, environment_metadata)
+        plugin_hooks = get_plugin_hooks(self._config.plugins, environment_metadata)
 
         self.__hooks_lock = ReadWriteLock()
         self.__hooks = self._config.hooks + plugin_hooks  # type: List[Hook]

ldclient/impl/async_big_segments.py

@@ -21,10 +21,9 @@
 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.
+    store to track its status. Call start() to begin the status 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
 
@@ -36,6 +35,11 @@ def __init__(self, config: AsyncBigSegmentsConfig):
         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)
+
+    def start(self):
+        """Starts the status polling task. Separated from __init__ so the manager
+        can be built without a running loop; the client calls this from start()."""
+        if self.__poll_task is not None:
             self.__poll_task.start()
 
     async def stop(self):

ldclient/impl/async_flag_tracker.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable
+from typing import Any, Callable, Optional
 
 from ldclient.context import Context
 from ldclient.impl.aio.concurrency import AsyncCallbackScheduler, AsyncLock
@@ -45,7 +45,14 @@ class AsyncFlagTrackerImpl(AsyncFlagTracker):
     def __init__(self, listeners: Listeners, eval_fn: Callable):
         self.__listeners = listeners
         self.__eval_fn = eval_fn
-        self.__scheduler = AsyncCallbackScheduler()
+        self.__scheduler: Optional[AsyncCallbackScheduler] = None
+
+    def _get_scheduler(self) -> AsyncCallbackScheduler:
+        """Creates the callback scheduler on first use. Called only from async
+        methods, so a running loop always exists for it to capture."""
+        if self.__scheduler is None:
+            self.__scheduler = AsyncCallbackScheduler()
+        return self.__scheduler
 
     def add_listener(self, listener: Callable[[FlagChange], None]):
         self.__listeners.add(listener)
@@ -54,7 +61,7 @@ def remove_listener(self, listener: Callable[[FlagChange], None]):
         self.__listeners.remove(listener)
 
     async def add_flag_value_change_listener(self, key: str, context: Context, fn: Callable[[FlagValueChange], None]) -> Callable[[FlagChange], None]:
-        listener = await AsyncFlagValueChangeListener.create(key, context, fn, self.__eval_fn, self.__scheduler)
+        listener = await AsyncFlagValueChangeListener.create(key, context, fn, self.__eval_fn, self._get_scheduler())
         self.add_listener(listener)
 
         return listener

ldclient/impl/client_common.py

@@ -10,15 +10,17 @@
 
 import hashlib
 import hmac
-from typing import List
+from typing import List, Sequence, Union
 
-from ldclient.config import Config, SdkIdentityConfig
+from ldclient.config import SdkIdentityConfig
 from ldclient.context import Context
-from ldclient.hook import Hook
+from ldclient.hook import AsyncHook, Hook
 from ldclient.impl.util import log
 from ldclient.plugin import (
     ApplicationMetadata,
+    AsyncPlugin,
     EnvironmentMetadata,
+    Plugin,
     SdkMetadata
 )
 from ldclient.version import VERSION
@@ -46,9 +48,9 @@ def get_environment_metadata(config: SdkIdentityConfig, sdk_name: str) -> Enviro
     )
 
 
-def get_plugin_hooks(config: Config, environment_metadata: EnvironmentMetadata) -> List[Hook]:
-    hooks = []
-    for plugin in config.plugins:
+def get_plugin_hooks(plugins: Sequence[Union[Plugin, AsyncPlugin]], environment_metadata: EnvironmentMetadata) -> List:
+    hooks: List = []
+    for plugin in plugins:
         try:
             hooks.extend(plugin.get_hooks(environment_metadata))
         except Exception as e:

ldclient/impl/datasystem/__init__.py

@@ -6,10 +6,14 @@
 from abc import abstractmethod
 from enum import Enum
 from threading import Event
-from typing import Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from ldclient.impl.aio.concurrency import AsyncEvent
 
 from ldclient.impl.listeners import Listeners
 from ldclient.interfaces import (
+    AsyncReadOnlyStore,
     DataSourceStatusProvider,
     DataStoreStatusProvider,
     FlagTracker,
@@ -143,6 +147,80 @@ def store(self) -> ReadOnlyStore:
         raise NotImplementedError
 
 
+class AsyncDataSystem(Protocol):
+    """
+    Async counterpart of :class:`DataSystem`: the same requirements, with the
+    data system's background work running as asyncio tasks.
+    """
+
+    @abstractmethod
+    def start(self, set_on_ready: "AsyncEvent"):
+        """
+        Starts the data system.
+
+        This method will return immediately. The provided event will be set when the system
+        has reached an initial state (either permanently failed, e.g. due to bad auth, or
+        succeeded)
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def stop(self):
+        """
+        Halts the data system. Should be called when the client is closed to stop any long running
+        operations.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def data_source_status_provider(self) -> DataSourceStatusProvider:
+        """
+        Returns an interface for tracking the status of the data source.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def data_store_status_provider(self) -> DataStoreStatusProvider:
+        """
+        Returns an interface for tracking the status of a persistent data store.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def flag_change_listeners(self) -> Listeners:
+        """
+        Returns the collection of listeners for flag change events.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def data_availability(self) -> DataAvailability:
+        """
+        Indicates what form of data is currently available.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def target_availability(self) -> DataAvailability:
+        """
+        Indicates the ideal form of data attainable given the current configuration.
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def store(self) -> AsyncReadOnlyStore:
+        """
+        Returns the data store used by the data system.
+        """
+        raise NotImplementedError
+
+
 class DiagnosticAccumulator(Protocol):
     def record_stream_init(self, timestamp, duration, failed):
         raise NotImplementedError

ldclient/impl/datasystem/async_fdv1.py

@@ -0,0 +1,172 @@
+from typing import Any, Callable, Optional
+
+from ldclient.async_config import AsyncConfig
+from ldclient.impl.aio.concurrency import AsyncEvent
+from ldclient.impl.aio.transport import AsyncHTTPTransport, AsyncSSEFactory
+from ldclient.impl.datasource.async_feature_requester import (
+    AsyncFeatureRequesterImpl
+)
+from ldclient.impl.datasource.async_polling import AsyncPollingUpdateProcessor
+from ldclient.impl.datasource.async_status import AsyncDataSourceUpdateSinkImpl
+from ldclient.impl.datasource.async_streaming import (
+    AsyncStreamingUpdateProcessor
+)
+from ldclient.impl.datasource.status import DataSourceStatusProviderImpl
+from ldclient.impl.datastore.status import (
+    DataStoreStatusProviderImpl,
+    DataStoreUpdateSinkImpl
+)
+from ldclient.impl.datasystem import (
+    AsyncDataSystem,
+    DataAvailability,
+    DiagnosticAccumulator
+)
+from ldclient.impl.listeners import Listeners
+from ldclient.impl.stubs import AsyncNullUpdateProcessor
+from ldclient.impl.util import log
+from ldclient.interfaces import (
+    AsyncFeatureStore,
+    AsyncReadOnlyStore,
+    AsyncUpdateProcessor,
+    DataSourceStatusProvider,
+    DataStoreStatusProvider
+)
+
+
+class AsyncFDv1(AsyncDataSystem):
+    """
+    AsyncFDv1 provides the v1 data source and store behavior through the
+    AsyncDataSystem interface. It is the async version of
+    :class:`ldclient.impl.datasystem.fdv1.FDv1`. Unlike the sync side, it uses
+    the feature store directly and does not wrap it for persistent-store status
+    monitoring.
+    """
+
+    def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, session_provider: Callable[[], Any]):
+        self._config = config
+        self._store = store
+        # The client creates the aiohttp session lazily inside the loop; the data
+        # source resolves it here when it builds its network processor at start().
+        self._session_provider = session_provider
+
+        # Set up data store status tracking (no store wrapper)
+        self._data_store_listeners = Listeners()
+        self._data_store_update_sink = DataStoreUpdateSinkImpl(
+            self._data_store_listeners
+        )
+        # The provider only calls the store's monitoring methods, which the async
+        # store also has, so the sync-typed signature is fine.
+        self._data_store_status_provider_impl = DataStoreStatusProviderImpl(
+            self._store, self._data_store_update_sink  # type: ignore[arg-type]
+        )
+
+        # Set up the data source status tracking and listeners
+        self._data_source_listeners = Listeners()
+        self._flag_change_listeners = Listeners()
+        self._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(
+            self._store,
+            self._data_source_listeners,
+            self._flag_change_listeners,
+        )
+        self._data_source_status_provider_impl = DataSourceStatusProviderImpl(
+            self._data_source_listeners, self._data_source_update_sink
+        )
+
+        # v1 processors read the sink from the config for status updates. The config
+        # attribute is typed as the sync sink, but the async sink has the same methods.
+        self._config._data_source_update_sink = self._data_source_update_sink  # type: ignore[assignment]
+
+        # Update processor created in start(), because it needs the ready event
+        self._update_processor: Optional[AsyncUpdateProcessor] = None
+
+        # Diagnostic accumulator provided by client for streaming metrics
+        self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None
+
+    def start(self, set_on_ready: AsyncEvent):
+        """
+        Starts the v1 update processor and returns immediately. The provided
+        event is set by the processor upon first successful initialization or
+        upon permanent failure.
+        """
+        update_processor = self._make_update_processor(
+            self._config, self._store, set_on_ready
+        )
+        self._update_processor = update_processor
+        update_processor.start()
+
+    async def stop(self):
+        if self._update_processor is not None:
+            await self._update_processor.stop()
+
+    @property
+    def store(self) -> AsyncReadOnlyStore:
+        return self._store
+
+    def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator):
+        """
+        Sets the diagnostic accumulator for streaming initialization metrics.
+        This should be called before start() to ensure metrics are collected.
+        """
+        self._diagnostic_accumulator = diagnostic_accumulator
+
+    @property
+    def data_source_status_provider(self) -> DataSourceStatusProvider:
+        return self._data_source_status_provider_impl
+
+    @property
+    def data_store_status_provider(self) -> DataStoreStatusProvider:
+        return self._data_store_status_provider_impl
+
+    @property
+    def flag_change_listeners(self) -> Listeners:
+        return self._flag_change_listeners
+
+    @property
+    def data_availability(self) -> DataAvailability:
+        if self._config.offline:
+            return DataAvailability.DEFAULTS
+
+        if self._update_processor is not None and self._update_processor.initialized():
+            return DataAvailability.REFRESHED
+
+        if self._store.initialized:
+            return DataAvailability.CACHED
+
+        return DataAvailability.DEFAULTS
+
+    @property
+    def target_availability(self) -> DataAvailability:
+        if self._config.offline:
+            return DataAvailability.DEFAULTS
+        # In LDD mode or normal connected modes, the ideal is to be refreshed
+        return DataAvailability.REFRESHED
+
+    def _make_update_processor(self, config: AsyncConfig, store: AsyncFeatureStore, ready: AsyncEvent):
+        # Mirrors FDv1._make_update_processor but builds the async processors
+        if config.update_processor_class:
+            log.info("Using user-specified update processor: " + str(config.update_processor_class))
+            return config.update_processor_class(config, store, ready)
+
+        if config.offline or config.use_ldd:
+            return AsyncNullUpdateProcessor(config, store, ready)
+
+        if config.stream:
+            return AsyncStreamingUpdateProcessor(
+                config,
+                store,
+                ready,
+                self._diagnostic_accumulator,
+                AsyncSSEFactory(config, session=self._session_provider(), proxy=config.http.http_proxy),
+            )
+
+        log.info("Disabling streaming API")
+        log.warning("You should only disable the streaming API if instructed to do so by LaunchDarkly support")
+
+        if config.feature_requester_class:
+            feature_requester = config.feature_requester_class(config)
+        else:
+            feature_requester = AsyncFeatureRequesterImpl(
+                config,
+                AsyncHTTPTransport(config, client=self._session_provider()),
+            )
+        return AsyncPollingUpdateProcessor(config, feature_requester, store, ready)