feat: Add data system builders for the async client

docs/api-main.rst

@@ -40,6 +40,13 @@ ldclient.datasystem module
     :members:
     :special-members: __init__
 
+ldclient.async_datasystem module
+--------------------------------
+
+.. automodule:: ldclient.async_datasystem
+    :members:
+    :special-members: __init__
+
 ldclient.hook module
 --------------------------
 

ldclient/async_datasystem.py

@@ -0,0 +1,186 @@
+"""
+Configuration for the async SDK's data acquisition strategy.
+
+This module is the async counterpart of :mod:`ldclient.datasystem`. Its functions build an
+:class:`ldclient.async_config.AsyncDataSystemConfig` for
+:class:`ldclient.async_client.AsyncLDClient`.
+
+.. 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 List, Optional
+
+from ldclient.async_config import AsyncDataSystemConfig
+from ldclient.config import DataSourceBuilder
+from ldclient.impl.datasourcev2.async_polling import (
+    AsyncFallbackToFDv1PollingDataSourceBuilder,
+    AsyncPollingDataSourceBuilder
+)
+from ldclient.impl.datasourcev2.async_streaming import (
+    AsyncStreamingDataSourceBuilder
+)
+from ldclient.interfaces import (
+    AsyncFeatureStore,
+    AsyncInitializer,
+    AsyncSynchronizer,
+    DataStoreMode
+)
+
+
+class AsyncConfigBuilder:  # pylint: disable=too-few-public-methods
+    """
+    Builder for the async data system configuration.
+    """
+
+    def __init__(self) -> None:
+        self._initializers: Optional[List[DataSourceBuilder[AsyncInitializer]]] = None
+        self._synchronizers: List[DataSourceBuilder[AsyncSynchronizer]] = []
+        self._fdv1_fallback_synchronizer: Optional[DataSourceBuilder[AsyncSynchronizer]] = None
+        self._store_mode: DataStoreMode = DataStoreMode.READ_ONLY
+        self._data_store: Optional[AsyncFeatureStore] = None
+
+    def initializers(self, initializers: Optional[List[DataSourceBuilder[AsyncInitializer]]]) -> "AsyncConfigBuilder":
+        """
+        Sets the initializers for the data system.
+        """
+        self._initializers = initializers
+        return self
+
+    def synchronizers(self, *sync_builders: DataSourceBuilder[AsyncSynchronizer]) -> "AsyncConfigBuilder":
+        """
+        Sets the synchronizers for the data system.
+
+        Accepts one or more synchronizer builders, ordered by preference. The first synchronizer is
+        the most preferred. The others are fallbacks, in order of decreasing preference.
+        """
+        if len(sync_builders) == 0:
+            raise ValueError("At least one synchronizer must be provided")
+        self._synchronizers = list(sync_builders)
+        return self
+
+    def fdv1_compatible_synchronizer(self, fallback: DataSourceBuilder[AsyncSynchronizer]) -> "AsyncConfigBuilder":
+        """
+        Configures the SDK with a fallback synchronizer that is compatible with the Flag Delivery
+        v1 API.
+        """
+        self._fdv1_fallback_synchronizer = fallback
+        return self
+
+    def data_store(self, data_store: AsyncFeatureStore, store_mode: DataStoreMode) -> "AsyncConfigBuilder":
+        """
+        Sets the async persistent data store and the mode in which the SDK uses it.
+        """
+        self._data_store = data_store
+        self._store_mode = store_mode
+        return self
+
+    def build(self) -> AsyncDataSystemConfig:
+        """
+        Builds the async data system configuration.
+        """
+        return AsyncDataSystemConfig(
+            initializers=self._initializers,
+            synchronizers=self._synchronizers if len(self._synchronizers) > 0 else None,
+            fdv1_fallback_synchronizer=self._fdv1_fallback_synchronizer,
+            data_store_mode=self._store_mode,
+            data_store=self._data_store,
+        )
+
+
+def polling_ds_builder() -> AsyncPollingDataSourceBuilder:
+    """
+    Returns a builder for an async polling data source.
+    """
+    return AsyncPollingDataSourceBuilder()
+
+
+def fdv1_fallback_ds_builder() -> AsyncFallbackToFDv1PollingDataSourceBuilder:
+    """
+    Returns a builder for an async fallback polling data source that is compatible with the Flag
+    Delivery v1 API.
+    """
+    return AsyncFallbackToFDv1PollingDataSourceBuilder()
+
+
+def streaming_ds_builder() -> AsyncStreamingDataSourceBuilder:
+    """
+    Returns a builder for an async streaming data source.
+    """
+    return AsyncStreamingDataSourceBuilder()
+
+
+def default() -> AsyncConfigBuilder:
+    """
+    Default is LaunchDarkly's recommended flag data acquisition strategy.
+
+    The SDK first requests data from LaunchDarkly's global CDN. Then it opens a streaming
+    connection to LaunchDarkly's Flag Delivery services to receive real-time updates. If the
+    streaming connection is interrupted for an extended period of time, the SDK falls back to
+    polling the global CDN for updates.
+    """
+    polling_builder = polling_ds_builder()
+    streaming_builder = streaming_ds_builder()
+    fallback = fdv1_fallback_ds_builder()
+
+    builder = AsyncConfigBuilder()
+    builder.initializers([polling_builder])
+    builder.synchronizers(streaming_builder, polling_builder)
+    builder.fdv1_compatible_synchronizer(fallback)
+
+    return builder
+
+
+def streaming() -> AsyncConfigBuilder:
+    """
+    Streaming configures the SDK to stream flag and segment data in the background, so that
+    evaluations use the latest data with no additional latency.
+    """
+    builder = AsyncConfigBuilder()
+    builder.synchronizers(streaming_ds_builder())
+    builder.fdv1_compatible_synchronizer(fdv1_fallback_ds_builder())
+
+    return builder
+
+
+def polling() -> AsyncConfigBuilder:
+    """
+    Polling configures the SDK to poll an endpoint for flag and segment data at a regular
+    interval. This is less efficient than streaming, but some network environments need it.
+    """
+    polling_builder: DataSourceBuilder[AsyncSynchronizer] = polling_ds_builder()
+
+    builder = AsyncConfigBuilder()
+    builder.synchronizers(polling_builder)
+    builder.fdv1_compatible_synchronizer(fdv1_fallback_ds_builder())
+
+    return builder
+
+
+def custom() -> AsyncConfigBuilder:
+    """
+    Custom returns an empty builder for a custom data acquisition strategy. You configure how the
+    SDK uses a persistent store, how it gets an initial set of data, and how it keeps the data up
+    to date.
+    """
+    return AsyncConfigBuilder()
+
+
+def daemon(store: AsyncFeatureStore) -> AsyncConfigBuilder:
+    """
+    Daemon configures the SDK to read from an async persistent store that the Relay Proxy or
+    another SDK populates. The SDK does not connect to LaunchDarkly, and it never writes to the
+    data store.
+    """
+    return custom().data_store(store, DataStoreMode.READ_ONLY)
+
+
+def persistent_store(store: AsyncFeatureStore) -> AsyncConfigBuilder:
+    """
+    PersistentStore is the same as :func:`default`, with an async persistent store added. Before
+    data arrives from LaunchDarkly, the SDK evaluates flags with data from the persistent store.
+    After fresh data arrives, the SDK no longer reads from the persistent store, but it keeps the
+    store up to date.
+    """
+    return default().data_store(store, DataStoreMode.READ_WRITE)

ldclient/testing/impl/datasystem/test_async_datasystem.py

@@ -0,0 +1,163 @@
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+from ldclient.async_client import AsyncLDClient
+from ldclient.async_config import AsyncConfig, AsyncDataSystemConfig
+from ldclient.async_datasystem import (
+    AsyncConfigBuilder,
+    custom,
+    daemon,
+    default,
+    fdv1_fallback_ds_builder,
+    persistent_store,
+    polling,
+    polling_ds_builder,
+    streaming,
+    streaming_ds_builder
+)
+from ldclient.async_feature_store import AsyncInMemoryFeatureStore
+from ldclient.context import Context
+from ldclient.impl.datasourcev2.async_polling import (
+    AsyncFallbackToFDv1PollingDataSourceBuilder,
+    AsyncPollingDataSourceBuilder
+)
+from ldclient.impl.datasourcev2.async_streaming import (
+    AsyncStreamingDataSourceBuilder
+)
+from ldclient.integrations.test_datav2 import TestDataV2
+from ldclient.interfaces import DataStoreMode
+
+
+def test_empty_builder_builds_an_empty_config():
+    config = AsyncConfigBuilder().build()
+
+    assert config == AsyncDataSystemConfig(
+        initializers=None,
+        synchronizers=None,
+        fdv1_fallback_synchronizer=None,
+        data_store_mode=DataStoreMode.READ_ONLY,
+        data_store=None,
+    )
+
+
+def test_builder_sets_every_field():
+    initializer = polling_ds_builder()
+    primary = streaming_ds_builder()
+    secondary = polling_ds_builder()
+    fallback = fdv1_fallback_ds_builder()
+    store = AsyncInMemoryFeatureStore()
+
+    config = (
+        AsyncConfigBuilder()
+        .initializers([initializer])
+        .synchronizers(primary, secondary)
+        .fdv1_compatible_synchronizer(fallback)
+        .data_store(store, DataStoreMode.READ_WRITE)
+        .build()
+    )
+
+    assert config.initializers == [initializer]
+    assert config.synchronizers == [primary, secondary]
+    assert config.fdv1_fallback_synchronizer is fallback
+    assert config.data_store is store
+    assert config.data_store_mode == DataStoreMode.READ_WRITE
+
+
+def test_synchronizers_needs_at_least_one_builder():
+    with pytest.raises(ValueError):
+        AsyncConfigBuilder().synchronizers()
+
+
+def test_custom_returns_a_new_empty_builder():
+    first = custom()
+    second = custom()
+
+    assert first is not second
+    assert first.build() == AsyncConfigBuilder().build()
+
+
+def test_default_polls_first_then_streams_with_polling_and_fdv1_fallbacks():
+    config = default().build()
+
+    assert len(config.initializers) == 1
+    assert isinstance(config.initializers[0], AsyncPollingDataSourceBuilder)
+    assert len(config.synchronizers) == 2
+    assert isinstance(config.synchronizers[0], AsyncStreamingDataSourceBuilder)
+    assert isinstance(config.synchronizers[1], AsyncPollingDataSourceBuilder)
+    assert isinstance(config.fdv1_fallback_synchronizer, AsyncFallbackToFDv1PollingDataSourceBuilder)
+    assert config.data_store is None
+
+
+def test_streaming_streams_with_an_fdv1_fallback():
+    config = streaming().build()
+
+    assert config.initializers is None
+    assert len(config.synchronizers) == 1
+    assert isinstance(config.synchronizers[0], AsyncStreamingDataSourceBuilder)
+    assert isinstance(config.fdv1_fallback_synchronizer, AsyncFallbackToFDv1PollingDataSourceBuilder)
+
+
+def test_polling_polls_with_an_fdv1_fallback():
+    config = polling().build()
+
+    assert config.initializers is None
+    assert len(config.synchronizers) == 1
+    assert isinstance(config.synchronizers[0], AsyncPollingDataSourceBuilder)
+    assert isinstance(config.fdv1_fallback_synchronizer, AsyncFallbackToFDv1PollingDataSourceBuilder)
+
+
+def test_daemon_reads_only_from_the_store():
+    store = AsyncInMemoryFeatureStore()
+    config = daemon(store).build()
+
+    assert config.initializers is None
+    assert config.synchronizers is None
+    assert config.fdv1_fallback_synchronizer is None
+    assert config.data_store is store
+    assert config.data_store_mode == DataStoreMode.READ_ONLY
+
+
+def test_persistent_store_is_the_default_with_a_writable_store():
+    store = AsyncInMemoryFeatureStore()
+    config = persistent_store(store).build()
+
+    assert isinstance(config.initializers[0], AsyncPollingDataSourceBuilder)
+    assert isinstance(config.synchronizers[0], AsyncStreamingDataSourceBuilder)
+    assert isinstance(config.fdv1_fallback_synchronizer, AsyncFallbackToFDv1PollingDataSourceBuilder)
+    assert config.data_store is store
+    assert config.data_store_mode == DataStoreMode.READ_WRITE
+
+
+def test_client_gives_its_session_to_every_builder_of_the_default_config():
+    # The async client shares one aiohttp session with each data source that the default
+    # strategy builds.
+    config = default().build()
+    session = object()
+    client = SimpleNamespace(_get_session=lambda: session)
+
+    with mock.patch.object(AsyncPollingDataSourceBuilder, 'session') as polling_session, \
+            mock.patch.object(AsyncStreamingDataSourceBuilder, 'session') as streaming_session, \
+            mock.patch.object(AsyncFallbackToFDv1PollingDataSourceBuilder, 'session') as fallback_session:
+        AsyncLDClient._wire_data_source_sessions(client, config)
+
+    # The polling builder is the initializer and also the second synchronizer.
+    assert polling_session.call_count == 2
+    assert streaming_session.call_count == 1
+    assert fallback_session.call_count == 1
+    for patched in (polling_session, streaming_session, fallback_session):
+        for call in patched.call_args_list:
+            assert call.args == (session,)
+
+
+@pytest.mark.asyncio
+async def test_client_evaluates_flags_from_a_custom_config():
+    td = TestDataV2.data_source()
+    td.update(td.flag('flag').on(True))
+    data_system = custom().synchronizers(td.async_builder).build()
+    config = AsyncConfig('sdk-key', send_events=False, datasystem_config=data_system)
+
+    async with AsyncLDClient(config) as client:
+        assert await client.is_initialized()
+        assert await client.variation('flag', Context.create('user'), False) is True