diff --git a/docs/api-main.rst b/docs/api-main.rst index 5b644b11..59d3ec41 100644 --- a/docs/api-main.rst +++ b/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 -------------------------- diff --git a/ldclient/async_datasystem.py b/ldclient/async_datasystem.py new file mode 100644 index 00000000..fd4a32cc --- /dev/null +++ b/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) diff --git a/ldclient/testing/impl/datasystem/test_async_datasystem.py b/ldclient/testing/impl/datasystem/test_async_datasystem.py new file mode 100644 index 00000000..859a8899 --- /dev/null +++ b/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