fix: Allow modifying fdv2 data source options independent of main config (#403)

Author: keelerm84

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,11 +52,10 @@ 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')
@@ -71,33 +71,34 @@ def __init__(self, tag, config):
                     streaming = primary.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()
+                        _set_optional_value(streaming, "baseUri", primary_builder.base_uri)
+                        _set_optional_time(streaming, "initialRetryDelayMs", primary_builder.initial_reconnect_delay)
                     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()
+                        _set_optional_value(polling, "baseUri", primary_builder.base_uri)
+                        _set_optional_time(polling, "pollIntervalMs", primary_builder.poll_interval)
+
                         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 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()
+                        _set_optional_value(streaming, "baseUri", secondary_builder.base_uri)
+                        _set_optional_time(streaming, "initialRetryDelayMs", secondary_builder.initial_reconnect_delay)
                     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()
+                        _set_optional_value(polling, "baseUri", secondary_builder.base_uri)
+                        _set_optional_time(polling, "pollIntervalMs", secondary_builder.poll_interval)
                         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)
@@ -307,7 +308,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,22 +156,35 @@ 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]]
+    primary_synchronizer: Optional[DataSourceBuilder[Synchronizer]]
     """The primary synchronizer for the data system."""
 
-    secondary_synchronizer: Optional[Builder[Synchronizer]] = None
+    secondary_synchronizer: Optional[DataSourceBuilder[Synchronizer]] = None
     """The secondary synchronizers for the data system."""
 
     data_store_mode: DataStoreMode = DataStoreMode.READ_WRITE
@@ -180,7 +193,7 @@ class DataSystemConfig:
     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 +454,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 +608,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,21 @@
     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._primary_synchronizer: Optional[DataSourceBuilder[Synchronizer]] = None
+        self._secondary_synchronizer: Optional[DataSourceBuilder[Synchronizer]] = None
+        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,8 +43,8 @@ def initializers(self, initializers: Optional[List[Builder[Initializer]]]) -> "C
 
     def synchronizers(
         self,
-        primary: Builder[Synchronizer],
-        secondary: Optional[Builder[Synchronizer]] = None,
+        primary: DataSourceBuilder[Synchronizer],
+        secondary: Optional[DataSourceBuilder[Synchronizer]] = None,
     ) -> "ConfigBuilder":
         """
         Sets the synchronizers for the data system.
@@ -65,7 +55,7 @@ def synchronizers(
 
     def fdv1_compatible_synchronizer(
             self,
-            fallback: Builder[Synchronizer]
+            fallback: DataSourceBuilder[Synchronizer]
     ) -> "ConfigBuilder":
         """
         Configures the SDK with a fallback synchronizer that is compatible with
@@ -99,40 +89,32 @@ def build(self) -> DataSystemConfig:
         )
 
 
-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 +167,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()