From 50c17e6e0c21cb1e8f2fcb134090411af66935d5 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 5 Nov 2026 10:56:16 -0500 Subject: [PATCH 1/2] fix: Stop sending the payload filter from the FDv2 data system The FDv2 backend does not support payload filtering. The FDv2 streaming and polling data sources, and the FDv1 fallback requesters that the FDv2 data system uses, no longer add the filter query parameter. The fallback thus cannot change the client to a filtered subset of the data. DataSourceBuilderConfig no longer has payload_filter_key, because only the FDv2 data sources read that protocol. Config and AsyncConfig keep the option for the FDv1 data system. --- contract-tests/async_client_entity.py | 8 +- contract-tests/client_entity.py | 2 - ldclient/config.py | 4 - ldclient/impl/datasourcev2/async_polling.py | 19 +-- ldclient/impl/datasourcev2/async_streaming.py | 3 - ldclient/impl/datasourcev2/polling.py | 19 +-- ldclient/impl/datasourcev2/streaming.py | 3 - .../impl/datasourcev2/test_payload_filter.py | 114 ++++++++++++++++++ 8 files changed, 121 insertions(+), 51 deletions(-) create mode 100644 ldclient/testing/impl/datasourcev2/test_payload_filter.py diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index 2569a156..fabeb847 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -301,9 +301,9 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy """Build an AsyncDataSystemConfig from the harness's dataSystem config. Wires the FDv2 initializers, the ordered synchronizer chain, the FDv1 - fallback synchronizer, the payload filter, and an optional async - persistent store. The async client injects its shared aiohttp session - into these builders when it starts. + fallback synchronizer, and an optional async persistent store. The async + client injects its shared aiohttp session into these builders when it + starts. """ initializers: Optional[list] = None init_configs = datasystem_config.get('initializers') @@ -347,8 +347,6 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) fdv1_fallback_synchronizer = fallback_builder - if datasystem_config.get("payloadFilter") is not None: - opts["payload_filter_key"] = datasystem_config["payloadFilter"] ds_kwargs: dict = { "initializers": initializers, diff --git a/contract-tests/client_entity.py b/contract-tests/client_entity.py index f93b5d85..7e41a1e4 100644 --- a/contract-tests/client_entity.py +++ b/contract-tests/client_entity.py @@ -91,8 +91,6 @@ def __init__(self, tag, config): _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) datasystem.fdv1_compatible_synchronizer(fallback_builder) - if datasystem_config.get("payloadFilter") is not None: - opts["payload_filter_key"] = datasystem_config["payloadFilter"] # Handle persistent data store configuration for dataSystem store_config = datasystem_config.get("store") diff --git a/ldclient/config.py b/ldclient/config.py index 81271ee8..b6a476a0 100644 --- a/ldclient/config.py +++ b/ldclient/config.py @@ -221,10 +221,6 @@ def initial_reconnect_delay(self) -> float: def poll_interval(self) -> float: """The interval between polling requests.""" - @property - def payload_filter_key(self) -> Optional[str]: - """The payload filter key applied as a query parameter.""" - # Set by the client after construction, so it is a read-write attribute # rather than a read-only property. _instance_id: Optional[str] diff --git a/ldclient/impl/datasourcev2/async_polling.py b/ldclient/impl/datasourcev2/async_polling.py index e3b6a3a3..38c90712 100644 --- a/ldclient/impl/datasourcev2/async_polling.py +++ b/ldclient/impl/datasourcev2/async_polling.py @@ -181,17 +181,9 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - - if selector is not None and selector.is_defined(): - query_params["basis"] = selector.state - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" + if selector is not None and selector.is_defined(): + uri += "?%s" % parse.urlencode({"basis": selector.state}) hdrs = _headers(self._config, ASYNC_USER_AGENT) hdrs["Accept-Encoding"] = "gzip" @@ -379,14 +371,7 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" hdrs = _headers(self._config, ASYNC_USER_AGENT) hdrs["Accept-Encoding"] = "gzip" diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 0f8428ab..4d117ea8 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -6,7 +6,6 @@ import json from time import time from typing import AsyncGenerator, Callable, Optional, Tuple -from urllib import parse import aiohttp from ld_eventsource import AsyncSSEClient @@ -70,8 +69,6 @@ def create_sse_client( caller owns it and ``None`` is returned in its place. """ uri = base_uri + STREAMING_ENDPOINT - if config.payload_filter_key is not None: - uri += "?%s" % parse.urlencode({"filter": config.payload_filter_key}) def query_params() -> dict: selector = ss.selector() diff --git a/ldclient/impl/datasourcev2/polling.py b/ldclient/impl/datasourcev2/polling.py index a96d5ed0..a7e1cf4d 100644 --- a/ldclient/impl/datasourcev2/polling.py +++ b/ldclient/impl/datasourcev2/polling.py @@ -160,17 +160,9 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - - if selector is not None and selector.is_defined(): - query_params["basis"] = selector.state - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" + if selector is not None and selector.is_defined(): + uri += "?%s" % parse.urlencode({"basis": selector.state}) hdrs = _headers(self._config) hdrs["Accept-Encoding"] = "gzip" @@ -334,14 +326,7 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" hdrs = _headers(self._config) hdrs["Accept-Encoding"] = "gzip" diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 5e9ab99a..f896d405 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -6,7 +6,6 @@ import json from time import time from typing import Callable, Generator, Optional, Tuple -from urllib import parse from ld_eventsource import SSEClient from ld_eventsource.actions import Event, Fault, Start @@ -73,8 +72,6 @@ def create_sse_client( to the LaunchDarkly streaming endpoint. """ uri = base_uri + STREAMING_ENDPOINT - if config.payload_filter_key is not None: - uri += "?%s" % parse.urlencode({"filter": config.payload_filter_key}) # We don't want the stream to use the same read timeout as the rest of the SDK. base_headers = _base_headers(config) diff --git a/ldclient/testing/impl/datasourcev2/test_payload_filter.py b/ldclient/testing/impl/datasourcev2/test_payload_filter.py new file mode 100644 index 00000000..471f02c7 --- /dev/null +++ b/ldclient/testing/impl/datasourcev2/test_payload_filter.py @@ -0,0 +1,114 @@ +""" +The FDv2 data system does not support payload filtering. These tests show that its data sources do +not send the payload filter key, also when the config has one. This includes the FDv1 fallback +requesters, so that a fallback does not change the client to a filtered subset of the data. +""" + +from unittest import mock +from urllib.parse import parse_qs, urlsplit + +import pytest + +from ldclient.config import Config +from ldclient.impl.datasourcev2 import async_streaming, streaming +from ldclient.impl.datasourcev2.async_polling import ( + AiohttpFDv1PollingRequester, + AiohttpPollingRequester +) +from ldclient.impl.datasourcev2.polling import ( + Urllib3FDv1PollingRequester, + Urllib3PollingRequester +) +from ldclient.interfaces import Selector +from ldclient.testing.http_util import start_server +from ldclient.testing.mock_components import MockSelectorStore + +FILTER_KEY = 'microservice-1' +SELECTOR = Selector.new_selector('p:SOMETHING:300', 300) + + +def path_and_query(request): + parts = urlsplit(request.path) + return parts.path, parse_qs(parts.query) + + +def make_config(base_uri: str = 'http://localhost') -> Config: + return Config('sdk-key', base_uri=base_uri, stream_uri=base_uri, payload_filter_key=FILTER_KEY) + + +@pytest.mark.parametrize('selector, expected_query', [ + (None, {}), + (SELECTOR, {'basis': [SELECTOR.state]}), +]) +def test_polling_requester_does_not_send_the_filter(selector, expected_query): + with start_server() as server: + config = make_config(server.uri) + requester = Urllib3PollingRequester(config, config.base_uri, config.http) + requester.fetch(selector) + assert path_and_query(server.require_request()) == ('/sdk/poll', expected_query) + + +def test_fdv1_fallback_polling_requester_does_not_send_the_filter(): + with start_server() as server: + config = make_config(server.uri) + requester = Urllib3FDv1PollingRequester(config, config.base_uri, config.http) + requester.fetch(SELECTOR) + assert path_and_query(server.require_request()) == ('/sdk/latest-all', {}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('selector, expected_query', [ + (None, {}), + (SELECTOR, {'basis': [SELECTOR.state]}), +]) +async def test_async_polling_requester_does_not_send_the_filter(selector, expected_query): + with start_server() as server: + config = make_config(server.uri) + requester = AiohttpPollingRequester(config, config.base_uri, config.http) + try: + await requester.fetch(selector) + finally: + await requester._http.close() + assert path_and_query(server.require_request()) == ('/sdk/poll', expected_query) + + +@pytest.mark.asyncio +async def test_async_fdv1_fallback_polling_requester_does_not_send_the_filter(): + with start_server() as server: + config = make_config(server.uri) + requester = AiohttpFDv1PollingRequester(config, config.base_uri, config.http) + try: + await requester.fetch(SELECTOR) + finally: + await requester._http.close() + assert path_and_query(server.require_request()) == ('/sdk/latest-all', {}) + + +@pytest.mark.parametrize('selector, expected_params', [ + (Selector.no_selector(), {}), + (SELECTOR, {'basis': SELECTOR.state}), +]) +def test_streaming_does_not_send_the_filter(selector, expected_params): + config = make_config() + with mock.patch.object(streaming.ConnectStrategy, 'http', wraps=streaming.ConnectStrategy.http) as http: + streaming.create_sse_client(config.stream_base_uri, config.http, config.initial_reconnect_delay, config, MockSelectorStore(selector)) + + kwargs = http.call_args.kwargs + assert kwargs['url'] == 'http://localhost/sdk/stream' + assert kwargs['query_params']() == expected_params + + +@pytest.mark.parametrize('selector, expected_params', [ + (Selector.no_selector(), {}), + (SELECTOR, {'basis': SELECTOR.state}), +]) +def test_async_streaming_does_not_send_the_filter(selector, expected_params): + config = make_config() + with mock.patch.object(async_streaming, 'AsyncSSEFactory') as factory_cls: + async_streaming.create_sse_client( + config.stream_base_uri, config.http, config.initial_reconnect_delay, config, MockSelectorStore(selector), session=object() + ) + + args, kwargs = factory_cls.return_value.create.call_args + assert args[0] == 'http://localhost/sdk/stream' + assert kwargs['query_params']() == expected_params From aaaba343724eb0aba7fb3c745515544eba3a6e5a Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 5 Nov 2026 11:18:16 -0500 Subject: [PATCH 2/2] chore: Remove extra blank lines from the contract-test services --- contract-tests/async_client_entity.py | 1 - contract-tests/client_entity.py | 1 - 2 files changed, 2 deletions(-) diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index fabeb847..87cd85a7 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -347,7 +347,6 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) fdv1_fallback_synchronizer = fallback_builder - ds_kwargs: dict = { "initializers": initializers, "synchronizers": synchronizers, diff --git a/contract-tests/client_entity.py b/contract-tests/client_entity.py index 7e41a1e4..b3544020 100644 --- a/contract-tests/client_entity.py +++ b/contract-tests/client_entity.py @@ -91,7 +91,6 @@ def __init__(self, tag, config): _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) datasystem.fdv1_compatible_synchronizer(fallback_builder) - # Handle persistent data store configuration for dataSystem store_config = datasystem_config.get("store") if store_config is not None: