Merge branch 'main' into jb/sdk-1748/minor-changes

Author: jsonbailey

.github/workflows/ci.yml

@@ -61,16 +61,16 @@ jobs:
         with:
           test_service_port: 9000
           token: ${{ secrets.GITHUB_TOKEN }}
-          stop_service: 'false'
-          enable_persistence_tests: 'true'
+          stop_service: "false"
+          enable_persistence_tests: "true"
 
       - name: Run contract tests v3
         uses: launchdarkly/gh-actions/actions/contract-tests@contract-tests-v1
         with:
           test_service_port: 9000
           token: ${{ secrets.GITHUB_TOKEN }}
-          version: v3.0.0-alpha.2
-          enable_persistence_tests: 'true'
+          version: v3.0.0-alpha.3
+          enable_persistence_tests: "true"
 
   windows:
     runs-on: windows-latest

contract-tests/client_entity.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import sys
+from typing import Any, Callable
 from urllib.parse import urlparse
 
 import requests
@@ -51,56 +52,38 @@ def __init__(self, tag, config):
                 for init_config in init_configs:
                     polling = init_config.get('polling')
                     if polling is not None:
-                        if polling.get("baseUri") is not None:
-                            opts["base_uri"] = polling["baseUri"]
-                        _set_optional_time_prop(polling, "pollIntervalMs", opts, "poll_interval")
-                        polling = polling_ds_builder()
-                        initializers.append(polling)
+                        polling_builder = polling_ds_builder()
+                        _set_optional_value(polling, "baseUri", polling_builder.base_uri)
+                        _set_optional_time(polling, "pollIntervalMs", polling_builder.poll_interval)
+                        initializers.append(polling_builder)
 
                 datasystem.initializers(initializers)
-            sync_config = datasystem_config.get('synchronizers')
-            if sync_config is not None:
-                primary = sync_config.get('primary')
-                secondary = sync_config.get('secondary')
-
-                primary_builder = None
-                secondary_builder = None
+            sync_configs = datasystem_config.get('synchronizers')
+            if sync_configs is not None:
+                sync_builders = []
                 fallback_builder = None
 
-                if primary is not None:
-                    streaming = primary.get('streaming')
+                for sync_config in sync_configs:
+                    streaming = sync_config.get('streaming')
                     if streaming is not None:
-                        primary_builder = streaming_ds_builder()
-                        if streaming.get("baseUri") is not None:
-                            opts["stream_uri"] = streaming["baseUri"]
-                        _set_optional_time_prop(streaming, "initialRetryDelayMs", opts, "initial_reconnect_delay")
-                        primary_builder = streaming_ds_builder()
-                    elif primary.get('polling') is not None:
-                        polling = primary.get('polling')
-                        if polling.get("baseUri") is not None:
-                            opts["base_uri"] = polling["baseUri"]
-                        _set_optional_time_prop(polling, "pollIntervalMs", opts, "poll_interval")
-                        primary_builder = polling_ds_builder()
-                        fallback_builder = fdv1_fallback_ds_builder()
+                        builder = streaming_ds_builder()
+                        _set_optional_value(streaming, "baseUri", builder.base_uri)
+                        _set_optional_time(streaming, "initialRetryDelayMs", builder.initial_reconnect_delay)
+                        sync_builders.append(builder)
+                    elif sync_config.get('polling') is not None:
+                        polling = sync_config.get('polling')
+
+                        builder = polling_ds_builder()
+                        _set_optional_value(polling, "baseUri", builder.base_uri)
+                        _set_optional_time(polling, "pollIntervalMs", builder.poll_interval)
+                        sync_builders.append(builder)
 
-                if secondary is not None:
-                    streaming = secondary.get('streaming')
-                    if streaming is not None:
-                        secondary_builder = streaming_ds_builder()
-                        if streaming.get("baseUri") is not None:
-                            opts["stream_uri"] = streaming["baseUri"]
-                        _set_optional_time_prop(streaming, "initialRetryDelayMs", opts, "initial_reconnect_delay")
-                        secondary_builder = streaming_ds_builder()
-                    elif secondary.get('polling') is not None:
-                        polling = secondary.get('polling')
-                        if polling.get("baseUri") is not None:
-                            opts["base_uri"] = polling["baseUri"]
-                        _set_optional_time_prop(polling, "pollIntervalMs", opts, "poll_interval")
-                        secondary_builder = polling_ds_builder()
                         fallback_builder = fdv1_fallback_ds_builder()
+                        _set_optional_value(polling, "baseUri", fallback_builder.base_uri)
+                        _set_optional_time(polling, "pollIntervalMs", fallback_builder.poll_interval)
 
-                if primary_builder is not None:
-                    datasystem.synchronizers(primary_builder, secondary_builder)
+                if sync_builders:
+                    datasystem.synchronizers(*sync_builders)
                 if fallback_builder is not None:
                     datasystem.fdv1_compatible_synchronizer(fallback_builder)
 
@@ -307,7 +290,16 @@ def close(self):
 def _set_optional_time_prop(params_in: dict, name_in: str, params_out: dict, name_out: str):
     if params_in.get(name_in) is not None:
         params_out[name_out] = params_in[name_in] / 1000.0
-    return None
+
+
+def _set_optional_time(params_in: dict, name_in: str, func: Callable[[float], Any]):
+    if params_in.get(name_in) is not None:
+        func(params_in[name_in] / 1000.0)
+
+
+def _set_optional_value(params_in: dict, name_in: str, func: Callable[[Any], Any]):
+    if params_in.get(name_in) is not None:
+        func(params_in[name_in])
 
 
 def _create_persistent_store(persistent_store_config: dict):

ldclient/config.py

@@ -6,7 +6,7 @@
 
 from dataclasses import dataclass
 from threading import Event
-from typing import Callable, List, Optional, Set, TypeVar
+from typing import Callable, List, Optional, Protocol, Set, TypeVar
 
 from ldclient.feature_store import InMemoryFeatureStore
 from ldclient.hook import Hook
@@ -92,11 +92,11 @@ def stale_after(self) -> float:
 
 
 class HTTPConfig:
-    """Advanced HTTP configuration options for the SDK client.
+    """Advanced HTTP configuration options for the SDK client / data sources.
 
     This class groups together HTTP/HTTPS-related configuration properties that rarely need to be changed.
     If you need to set these, construct an ``HTTPConfig`` instance and pass it as the ``http`` parameter when
-    you construct the main :class:`Config` for the SDK client.
+    you construct the main :class:`Config` for the SDK client, or to the appropriate data source builder method.
     """
 
     def __init__(
@@ -156,31 +156,45 @@ def disable_ssl_verification(self) -> bool:
         return self.__disable_ssl_verification
 
 
-T = TypeVar("T")
+T_co = TypeVar("T_co", covariant=True)
 
-Builder = Callable[['Config'], T]
+
+class DataSourceBuilder(Protocol[T_co]):  # pylint: disable=too-few-public-methods
+    """
+    Protocol for building data sources.
+    """
+
+    def build(self, config: 'Config') -> T_co:
+        """
+        Builds the data source.
+
+        :param config: the SDK configuration
+        :return: the built data source
+        """
+        raise NotImplementedError
 
 
 @dataclass(frozen=True)
 class DataSystemConfig:
     """Configuration for LaunchDarkly's data acquisition strategy."""
 
-    initializers: Optional[List[Builder[Initializer]]]
+    initializers: Optional[List[DataSourceBuilder[Initializer]]]
     """The initializers for the data system."""
 
-    primary_synchronizer: Optional[Builder[Synchronizer]]
-    """The primary synchronizer for the data system."""
-
-    secondary_synchronizer: Optional[Builder[Synchronizer]] = None
-    """The secondary synchronizers for the data system."""
+    synchronizers: Optional[List[DataSourceBuilder[Synchronizer]]]
+    """
+    The synchronizers for the data system, ordered by preference.
+    The first synchronizer is the most preferred, with subsequent synchronizers
+    serving as fallbacks in order of decreasing preference.
+    """
 
     data_store_mode: DataStoreMode = DataStoreMode.READ_WRITE
     """The data store mode specifies the mode in which the persistent store will operate, if present."""
 
     data_store: Optional[FeatureStore] = None
     """The (optional) persistent data store instance."""
 
-    fdv1_fallback_synchronizer: Optional[Builder[Synchronizer]] = None
+    fdv1_fallback_synchronizer: Optional[DataSourceBuilder[Synchronizer]] = None
     """An optional fallback synchronizer that will read from FDv1"""
 
 
@@ -441,7 +455,7 @@ def event_processor_class(self) -> Optional[Callable[['Config'], EventProcessor]
         return self.__event_processor_class
 
     @property
-    def feature_requester_class(self) -> Callable:
+    def feature_requester_class(self) -> Optional[Callable]:
         return self.__feature_requester_class
 
     @property
@@ -595,4 +609,4 @@ def _validate(self):
             log.warning("Missing or blank SDK key")
 
 
-__all__ = ['Config', 'BigSegmentsConfig', 'DataSystemConfig', 'HTTPConfig']
+__all__ = ['Config', 'BigSegmentsConfig', 'DataSourceBuilder', 'DataSystemConfig', 'HTTPConfig']

ldclient/datasystem.py

@@ -2,22 +2,16 @@
 Configuration for LaunchDarkly's data acquisition strategy.
 """
 
-from typing import Callable, List, Optional, TypeVar
+from typing import List, Optional
 
-from ldclient.config import Config as LDConfig
-from ldclient.config import DataSystemConfig
+from ldclient.config import DataSourceBuilder, DataSystemConfig
 from ldclient.impl.datasourcev2.polling import (
-    PollingDataSource,
-    PollingDataSourceBuilder,
-    Urllib3FDv1PollingRequester,
-    Urllib3PollingRequester
-)
-from ldclient.impl.datasourcev2.streaming import (
-    StreamingDataSource,
-    StreamingDataSourceBuilder
+    FallbackToFDv1PollingDataSourceBuilder,
+    PollingDataSourceBuilder
 )
+from ldclient.impl.datasourcev2.streaming import StreamingDataSourceBuilder
 from ldclient.impl.integrations.files.file_data_sourcev2 import (
-    _FileDataSourceV2
+    FileDataSourceV2Builder
 )
 from ldclient.interfaces import (
     DataStoreMode,
@@ -26,25 +20,20 @@
     Synchronizer
 )
 
-T = TypeVar("T")
-
-Builder = Callable[[LDConfig], T]
-
 
 class ConfigBuilder:  # pylint: disable=too-few-public-methods
     """
     Builder for the data system configuration.
     """
 
     def __init__(self) -> None:
-        self._initializers: Optional[List[Builder[Initializer]]] = None
-        self._primary_synchronizer: Optional[Builder[Synchronizer]] = None
-        self._secondary_synchronizer: Optional[Builder[Synchronizer]] = None
-        self._fdv1_fallback_synchronizer: Optional[Builder[Synchronizer]] = None
+        self._initializers: Optional[List[DataSourceBuilder[Initializer]]] = None
+        self._synchronizers: List[DataSourceBuilder[Synchronizer]] = []
+        self._fdv1_fallback_synchronizer: Optional[DataSourceBuilder[Synchronizer]] = None
         self._store_mode: DataStoreMode = DataStoreMode.READ_ONLY
         self._data_store: Optional[FeatureStore] = None
 
-    def initializers(self, initializers: Optional[List[Builder[Initializer]]]) -> "ConfigBuilder":
+    def initializers(self, initializers: Optional[List[DataSourceBuilder[Initializer]]]) -> "ConfigBuilder":
         """
         Sets the initializers for the data system.
         """
@@ -53,19 +42,28 @@ def initializers(self, initializers: Optional[List[Builder[Initializer]]]) -> "C
 
     def synchronizers(
         self,
-        primary: Builder[Synchronizer],
-        secondary: Optional[Builder[Synchronizer]] = None,
+        *sync_builders: DataSourceBuilder[Synchronizer]
     ) -> "ConfigBuilder":
         """
         Sets the synchronizers for the data system.
+
+        Accepts one or more synchronizer builders, ordered by preference.
+        The first synchronizer is the most preferred, with subsequent
+        synchronizers serving as fallbacks in order of decreasing preference.
+
+        Examples:
+            builder.synchronizers(streaming_builder)
+            builder.synchronizers(streaming_builder, polling_builder)
+            builder.synchronizers(sync1, sync2, sync3)
         """
-        self._primary_synchronizer = primary
-        self._secondary_synchronizer = secondary
+        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: Builder[Synchronizer]
+            fallback: DataSourceBuilder[Synchronizer]
     ) -> "ConfigBuilder":
         """
         Configures the SDK with a fallback synchronizer that is compatible with
@@ -86,53 +84,41 @@ def build(self) -> DataSystemConfig:
         """
         Builds the data system configuration.
         """
-        if self._secondary_synchronizer is not None and self._primary_synchronizer is None:
-            raise ValueError("Primary synchronizer must be set if secondary is set")
-
         return DataSystemConfig(
             initializers=self._initializers,
-            primary_synchronizer=self._primary_synchronizer,
-            secondary_synchronizer=self._secondary_synchronizer,
+            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() -> Builder[PollingDataSource]:
-    def builder(config: LDConfig) -> PollingDataSource:
-        requester = Urllib3PollingRequester(config)
-        polling_ds = PollingDataSourceBuilder(config)
-        polling_ds.requester(requester)
-
-        return polling_ds.build()
-
-    return builder
-
-
-def fdv1_fallback_ds_builder() -> Builder[PollingDataSource]:
-    def builder(config: LDConfig) -> PollingDataSource:
-        requester = Urllib3FDv1PollingRequester(config)
-        polling_ds = PollingDataSourceBuilder(config)
-        polling_ds.requester(requester)
-
-        return polling_ds.build()
-
-    return builder
+def polling_ds_builder() -> PollingDataSourceBuilder:
+    """
+    Returns a builder for a polling data source.
+    """
+    return PollingDataSourceBuilder()
 
 
-def streaming_ds_builder() -> Builder[StreamingDataSource]:
-    def builder(config: LDConfig) -> StreamingDataSource:
-        return StreamingDataSourceBuilder(config).build()
+def fdv1_fallback_ds_builder() -> FallbackToFDv1PollingDataSourceBuilder:
+    """
+    Returns a builder for a Flag Delivery v1 compatible fallback polling data source.
+    """
+    return FallbackToFDv1PollingDataSourceBuilder()
 
-    return builder
 
+def streaming_ds_builder() -> StreamingDataSourceBuilder:
+    """
+    Returns a builder for a streaming data source.
+    """
+    return StreamingDataSourceBuilder()
 
-def file_ds_builder(paths: List[str]) -> Builder[Initializer]:
-    def builder(_: LDConfig) -> Initializer:
-        return _FileDataSourceV2(paths)
 
-    return builder
+def file_ds_builder(paths: List[str]) -> FileDataSourceV2Builder:
+    """
+    Returns a builder for a file-based data source.
+    """
+    return FileDataSourceV2Builder(paths)
 
 
 def default() -> ConfigBuilder:
@@ -185,7 +171,7 @@ def polling() -> ConfigBuilder:
     streaming, but may be necessary in some network environments.
     """
 
-    polling_builder: Builder[Synchronizer] = polling_ds_builder()
+    polling_builder: DataSourceBuilder[Synchronizer] = polling_ds_builder()
     fallback = fdv1_fallback_ds_builder()
 
     builder = ConfigBuilder()