refactor: Build client components in __init__ and create the session lazily

Construct the data system, status providers, big-segment manager, evaluator,
and flag tracker in __init__ (loop-free); start() only does the loop-bound work.
The shared aiohttp session is created lazily on first use inside the loop (via a
provider the data source resolves), so offline/LDD mode creates none. The
big-segment manager's poll task is created in __init__ and started in start().
This matches the loop-free-init / loop-bound-start shape used across the async
ecosystem, and makes the status providers available before start() too.

Author: jsonbailey

ldclient/async_client.py

@@ -37,7 +37,6 @@
     create_diagnostic_id
 )
 from ldclient.impl.events.types import EventFactory
-from ldclient.impl.listeners import Listeners
 from ldclient.impl.model.feature_flag import FeatureFlag
 from ldclient.impl.stubs import AsyncNullEventProcessor
 from ldclient.impl.util import log
@@ -53,18 +52,6 @@
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 
-class _NotStartedDataSystem:
-    """Placeholder data system used before start(); reports that only
-    application-provided defaults are available."""
-
-    @property
-    def data_availability(self) -> DataAvailability:
-        return DataAvailability.DEFAULTS
-
-    async def stop(self) -> None:
-        pass
-
-
 class AsyncLDClient:
     """Async LaunchDarkly SDK client.
 
@@ -96,23 +83,47 @@ def __init__(self, config: AsyncConfig):
 
         self._session = None
         self._proxy: Optional[str] = None
-        # Pre-start placeholders so that evaluation/track/identify before
-        # start() degrade gracefully (defaults returned, events dropped).
+        # Event processor is a no-op until start(); track/identify before start()
+        # drop events.
         self._event_processor: Any = AsyncNullEventProcessor()
-        self._data_system: AsyncDataSystem = _NotStartedDataSystem()  # type: ignore[assignment]
 
         self.__hooks: List = list(config.hooks)
 
         self._event_factory_default = EventFactory(False)
         self._event_factory_with_reasons = EventFactory(True)
 
-        self._flag_change_listeners = Listeners()
+        # Build the object graph here (loop-free). start() supplies the loop-bound
+        # resources: the HTTP session (created lazily), the data source, the
+        # big-segment poll, and the event processor. Evaluation before start()
+        # serves whatever the store already has.
+        self._data_system: AsyncDataSystem = self._make_data_system()
+
+        self.__data_store_status_provider = self._data_system.data_store_status_provider
+        self.__data_source_status_provider = self._data_system.data_source_status_provider
+
+        self.__big_segment_store_manager = AsyncBigSegmentStoreManager(self._config.big_segments)
+
+        async def get_flag_fn(key):
+            return await self._data_system.store.get(FEATURES, key)
+
+        async def get_segment_fn(key):
+            return await self._data_system.store.get(SEGMENTS, key)
+
+        async def get_membership_fn(key):
+            return await self.__big_segment_store_manager.get_user_membership(key)
+
+        self._evaluator = AsyncEvaluator(
+            get_flag_fn,
+            get_segment_fn,
+            get_membership_fn,
+            log,
+        )
 
         async def variation_eval_fn(key, context):
             return await self.variation(key, context, None)
 
         self.__flag_tracker = AsyncFlagTrackerImpl(
-            self._flag_change_listeners,
+            self._data_system.flag_change_listeners,
             variation_eval_fn
         )
 
@@ -208,33 +219,8 @@ async def __start_up(self, start_wait: float):
 
         self.__hooks = self._config.hooks + plugin_hooks
 
-        self._session = await self._create_http_session()
-        self._data_system = self._make_data_system()
-
-        # Expose providers and store from data system
-        self.__data_store_status_provider = self._data_system.data_store_status_provider
-        self.__data_source_status_provider = (
-            self._data_system.data_source_status_provider
-        )
-
-        big_segment_store_manager = AsyncBigSegmentStoreManager(self._config.big_segments)
-        self.__big_segment_store_manager = big_segment_store_manager
-
-        async def get_flag_fn(key):
-            return await self._data_system.store.get(FEATURES, key)
-
-        async def get_segment_fn(key):
-            return await self._data_system.store.get(SEGMENTS, key)
-
-        async def get_membership_fn(key):
-            return await big_segment_store_manager.get_user_membership(key)
-
-        self._evaluator = AsyncEvaluator(
-            get_flag_fn,
-            get_segment_fn,
-            get_membership_fn,
-            log,
-        )
+        # Start the big-segment status poll now that a loop is running.
+        self.__big_segment_store_manager.start()
 
         if self._config.offline:
             log.info("Started LaunchDarkly Client in offline mode")
@@ -265,8 +251,16 @@ async def get_membership_fn(key):
         else:
             log.warning("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. " "Feature Flags may not yet be available.")
 
-    async def _create_http_session(self):
-        """Create and return the aiohttp session. Called from __start_up."""
+    def _get_session(self):
+        """Return the shared aiohttp session, creating it on first use inside the
+        event loop. Nothing creates it in offline/LDD mode, because no network
+        component asks for it."""
+        if self._session is None:
+            self._session = self._create_http_session()
+        return self._session
+
+    def _create_http_session(self):
+        """Create and return the aiohttp session."""
         import ssl
 
         import aiohttp
@@ -293,7 +287,7 @@ def _make_data_system(self) -> AsyncDataSystem:
         if datasystem_config is None:
             from ldclient.impl.datasystem.async_fdv1 import AsyncFDv1
 
-            return AsyncFDv1(self._config, self._select_feature_store(), self._flag_change_listeners, self._session, self._proxy)
+            return AsyncFDv1(self._config, self._select_feature_store(), self._get_session)
 
         raise NotImplementedError("FDv2 is not yet supported in the async client")
 
@@ -319,7 +313,7 @@ def _set_event_processor(self, config):
         if not config.event_processor_class:
             diagnostic_id = create_diagnostic_id(config)
             diagnostic_accumulator = None if config.diagnostic_opt_out else _DiagnosticAccumulator(diagnostic_id)
-            self._event_processor = DefaultAsyncEventProcessor(config, self._session, diagnostic_accumulator=diagnostic_accumulator)
+            self._event_processor = DefaultAsyncEventProcessor(config, self._get_session(), diagnostic_accumulator=diagnostic_accumulator)
             return diagnostic_accumulator
         self._event_processor = config.event_processor_class(config)
         return None

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/datasystem/async_fdv1.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from ldclient.async_config import AsyncConfig
 from ldclient.impl.aio.concurrency import AsyncEvent
@@ -42,11 +42,12 @@ class AsyncFDv1(AsyncDataSystem):
     monitoring.
     """
 
-    def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, flag_change_listeners: Listeners, session: Optional[Any] = None, proxy: Optional[str] = None):
+    def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, session_provider: Callable[[], Any]):
         self._config = config
         self._store = store
-        self._session = session
-        self._proxy = proxy
+        # 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()
@@ -59,11 +60,9 @@ def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, flag_change_li
             self._store, self._data_store_update_sink  # type: ignore[arg-type]
         )
 
-        # Set up the data source status tracking and listeners. The flag-change
-        # Listeners is provided by the client so its flag tracker (built before
-        # start()) shares the same collection.
+        # Set up the data source status tracking and listeners
         self._data_source_listeners = Listeners()
-        self._flag_change_listeners = flag_change_listeners
+        self._flag_change_listeners = Listeners()
         self._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(
             self._store,
             self._data_source_listeners,
@@ -157,7 +156,7 @@ def _make_update_processor(self, config: AsyncConfig, store: AsyncFeatureStore,
                 store,
                 ready,
                 self._diagnostic_accumulator,
-                AsyncSSEFactory(config, session=self._session, proxy=self._proxy),
+                AsyncSSEFactory(config, session=self._session_provider(), proxy=config.http.http_proxy),
             )
 
         log.info("Disabling streaming API")
@@ -168,6 +167,6 @@ def _make_update_processor(self, config: AsyncConfig, store: AsyncFeatureStore,
         else:
             feature_requester = AsyncFeatureRequesterImpl(
                 config,
-                AsyncHTTPTransport(config, client=self._session),
+                AsyncHTTPTransport(config, client=self._session_provider()),
             )
         return AsyncPollingUpdateProcessor(config, feature_requester, store, ready)

ldclient/testing/impl/test_async_big_segments.py

@@ -64,8 +64,10 @@ def membership_queries(self):
 
 async def make_started_manager(store, **kwargs):
     config = AsyncBigSegmentsConfig(store=store, **kwargs)
-    # The constructor starts the polling task (it requires a running event loop).
-    return AsyncBigSegmentStoreManager(config)
+    manager = AsyncBigSegmentStoreManager(config)
+    # start() begins the polling task (it requires a running event loop).
+    manager.start()
+    return manager
 
 
 @pytest.mark.asyncio

ldclient/testing/test_async_client.py

@@ -223,16 +223,17 @@ def after_evaluation(self, series_context, data, detail):
 
 @pytest.mark.asyncio
 async def test_flag_tracker_available_before_start():
-    """flag_tracker is available before start(); the client's flag-change
-    Listeners is the same collection the data system uses after start()."""
+    """flag_tracker is available before start(); the data system it reads from is
+    built in __init__ and not rebuilt by start(), so early listeners stay wired."""
     client = AsyncLDClient(_offline_config())
 
     tracker = client.flag_tracker
     assert tracker is not None
     tracker.add_listener(lambda change: None)
 
+    data_system_before = client._data_system
     await client.start()
-    assert client._data_system.flag_change_listeners is client._flag_change_listeners
+    assert client._data_system is data_system_before
     await client.close()