From 3cef7b5df965347d33f4f9f254d863854dd1dc29 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 5 Aug 2026 16:23:54 -0600 Subject: [PATCH 1/4] feat: Add AsyncLDClient with FDv1 data system and public API --- ldclient/__init__.py | 9 +- ldclient/async_client.py | 772 +++++++++++++++++++++ ldclient/client.py | 2 +- ldclient/impl/client_common.py | 14 +- ldclient/impl/datasystem/__init__.py | 80 ++- ldclient/impl/datasystem/async_fdv1.py | 171 +++++ ldclient/impl/stubs.py | 49 +- ldclient/testing/mock_async_components.py | 11 +- ldclient/testing/stub_util.py | 5 + ldclient/testing/test_async_client.py | 364 ++++++++++ ldclient/testing/test_sync_async_parity.py | 73 ++ 11 files changed, 1536 insertions(+), 14 deletions(-) create mode 100644 ldclient/async_client.py create mode 100644 ldclient/impl/datasystem/async_fdv1.py create mode 100644 ldclient/testing/test_async_client.py create mode 100644 ldclient/testing/test_sync_async_parity.py diff --git a/ldclient/__init__.py b/ldclient/__init__.py index 1536f331..978d488a 100644 --- a/ldclient/__init__.py +++ b/ldclient/__init__.py @@ -88,4 +88,11 @@ def _reset_client(): __BASE_TYPES__ = (str, float, int, bool) -__all__ = ['Config', 'Context', 'ContextBuilder', 'ContextMultiBuilder', 'LDClient', 'Result', 'client', 'context', 'evaluation', 'integrations', 'interfaces', 'migrations'] +def __getattr__(name): + if name == 'AsyncLDClient': + from ldclient.async_client import AsyncLDClient + return AsyncLDClient + raise AttributeError("module 'ldclient' has no attribute %r" % name) + + +__all__ = ['AsyncLDClient', 'Config', 'Context', 'ContextBuilder', 'ContextMultiBuilder', 'LDClient', 'Result', 'client', 'context', 'evaluation', 'integrations', 'interfaces', 'migrations'] diff --git a/ldclient/async_client.py b/ldclient/async_client.py new file mode 100644 index 00000000..753441df --- /dev/null +++ b/ldclient/async_client.py @@ -0,0 +1,772 @@ +""" +Async client for the LaunchDarkly Server-Side Python SDK. +""" + +import asyncio +import traceback +from typing import Any, Callable, List, Optional, Tuple +from uuid import uuid4 + +import certifi + +from ldclient.async_config import AsyncConfig +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.context import Context +from ldclient.evaluation import EvaluationDetail, FeatureFlagsState +from ldclient.hook import ( + AsyncHook, + EvaluationSeriesContext, + _EvaluationWithHookResult +) +from ldclient.impl import AnyNum +from ldclient.impl.aio.concurrency import AsyncEvent +from ldclient.impl.async_big_segments import AsyncBigSegmentStoreManager +from ldclient.impl.async_evaluator import AsyncEvaluator, error_reason +from ldclient.impl.async_flag_tracker import AsyncFlagTrackerImpl +from ldclient.impl.client_common import ( + get_environment_metadata, + get_plugin_hooks +) +from ldclient.impl.client_common import secure_mode_hash as _secure_mode_hash +from ldclient.impl.datasystem import AsyncDataSystem, DataAvailability +from ldclient.impl.events.async_event_processor import ( + DefaultAsyncEventProcessor +) +from ldclient.impl.events.diagnostics import ( + _DiagnosticAccumulator, + create_diagnostic_id +) +from ldclient.impl.events.types import EventFactory +from ldclient.impl.model.feature_flag import FeatureFlag +from ldclient.impl.rwlock import ReadWriteLock +from ldclient.impl.stubs import AsyncNullEventProcessor +from ldclient.impl.util import log +from ldclient.interfaces import ( + AsyncFeatureStore, + AsyncFlagTracker, + BigSegmentStoreStatusProvider, + DataSourceStatusProvider, + DataStoreStatusProvider +) +from ldclient.migrations import OpTracker, Stage +from ldclient.plugin import EnvironmentMetadata +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +async def _get_store_item(store, kind: VersionedDataKind, key: str) -> Any: + # This decorator around store.get provides backward compatibility with any custom data + # store implementation that might still be returning a dict, instead of our data model + # classes like FeatureFlag. + item = await store.get(kind, key) + return kind.decode(item) if isinstance(item, dict) else item + + +class _NotStartedDataSystem: + """Placeholder data system used before start(); reports that only + application-provided defaults are available.""" + + @property + def data_availability(self) -> DataAvailability: + return DataAvailability.DEFAULTS + + async def stop(self) -> None: + pass + + +class AsyncLDClient: + """Async LaunchDarkly SDK client. + + .. 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. Pin to a specific minor version and review the changelog + before upgrading. + + Use ``async with AsyncLDClient(config) as client:`` or call + ``await client.start()`` / ``await client.close()`` explicitly. + """ + + def __init__(self, config: AsyncConfig): + """ + Construct an AsyncLDClient. Does NOT start background tasks; call + ``await start()`` (or use the async context manager) before evaluating flags. + + :param config: SDK configuration + """ + config._validate() + + self._config = config + self._config._instance_id = str(uuid4()) + self._lifecycle_lock = asyncio.Lock() + + self._started = False + self._closed = False + + self._session = None + self._proxy: Optional[str] = None + # Pre-start placeholders so that evaluation/track/identify before + # start() degrade gracefully (defaults returned, events dropped). + self._event_processor: Any = AsyncNullEventProcessor() + self._data_system: AsyncDataSystem = _NotStartedDataSystem() # type: ignore[assignment] + + self.__hooks_lock = ReadWriteLock() + self.__hooks: List = list(config.hooks) + + self._event_factory_default = EventFactory(False) + self._event_factory_with_reasons = EventFactory(True) + + async def start(self, start_wait: float = 5.0) -> None: + """Start the client: create the HTTP session, data system, and event processor. + + Safe to call multiple times — subsequent calls are no-ops. + + :param start_wait: seconds to wait for the data source to initialize + """ + async with self._lifecycle_lock: + if self._closed: + raise RuntimeError("Cannot start a closed AsyncLDClient") + if self._started: + return + + # __start_up resets the hook list to config.hooks + plugin hooks; + # preserve any hooks registered via add_hook() before start(). + with self.__hooks_lock.read(): + pre_start_hooks = [h for h in self.__hooks if h not in self._config.hooks] + + try: + await self.__start_up(start_wait) + self._started = True + except Exception: + await self._cleanup_partial_start() + raise + + for hook in pre_start_hooks: + self.add_hook(hook) + + async def _cleanup_partial_start(self): + """Release any resources that were partially created during a failed __start_up.""" + try: + await self._event_processor.stop() + except Exception: + pass + try: + await self._data_system.stop() + except Exception: + pass + try: + manager = self.__big_segment_store_manager + except AttributeError: + manager = None + if manager is not None: + try: + await manager.stop() + except Exception: + pass + if self._session is not None: + try: + await self._session.close() + except Exception: + pass + + async def close(self, close_timeout: float = 2.0) -> None: + """Shut down the client and release all resources. + + Safe to call multiple times — subsequent calls are no-ops. + """ + async with self._lifecycle_lock: + if self._closed: + return + self._closed = True + + if self._started: + try: + await asyncio.wait_for(self._close_components(), timeout=close_timeout) + except asyncio.TimeoutError: + log.warning("Timed out closing AsyncLDClient components") + except Exception as e: + log.warning("Error closing AsyncLDClient components: %s", e) + + # Close HTTP session + if self._session is not None: + try: + await self._session.close() + except Exception as e: + log.warning("Error closing HTTP session: %s", e) + + async def _close_components(self): + """Releases the threads and network connections used by the SDK + components. The public :meth:`close` wraps this with a timeout.""" + log.info("Closing LaunchDarkly client..") + await self._event_processor.stop() + await self._data_system.stop() + await self.__big_segment_store_manager.stop() + + async def __start_up(self, start_wait: float): + environment_metadata = get_environment_metadata(self._config, "python-server-sdk-async") + plugin_hooks = get_plugin_hooks(self._config.plugins, environment_metadata) + + self.__hooks_lock = ReadWriteLock() + self.__hooks = self._config.hooks + plugin_hooks + + self._session = await self._create_http_session() + self._data_system = self._make_data_system() + + async def variation_eval_fn(key, context): + return await self.variation(key, context, None) + + self.__flag_tracker = AsyncFlagTrackerImpl( + self._data_system.flag_change_listeners, + variation_eval_fn + ) + # Expose providers and store from data system + self.__data_store_status_provider = self._data_system.data_store_status_provider + self.__data_source_status_provider = ( + self._data_system.data_source_status_provider + ) + + big_segment_store_manager = AsyncBigSegmentStoreManager(self._config.big_segments) + self.__big_segment_store_manager = big_segment_store_manager + + async def get_flag_fn(key): + return await _get_store_item(self._data_system.store, FEATURES, key) + + async def get_segment_fn(key): + return await _get_store_item(self._data_system.store, SEGMENTS, key) + + async def get_membership_fn(key): + return await big_segment_store_manager.get_user_membership(key) + + self._evaluator = AsyncEvaluator( + get_flag_fn, + get_segment_fn, + get_membership_fn, + log, + ) + + if self._config.offline: + log.info("Started LaunchDarkly Client in offline mode") + + if self._config.use_ldd: + log.info("Started LaunchDarkly Client in LDD mode") + + diagnostic_accumulator = self._set_event_processor(self._config) + + # Pass diagnostic accumulator to data system for streaming metrics + self._data_system.set_diagnostic_accumulator(diagnostic_accumulator) # type: ignore + + await self.__register_plugins(environment_metadata) + + update_processor_ready = AsyncEvent() + self._data_system.start(update_processor_ready) + + if not self._config.offline and not self._config.use_ldd: + if start_wait > 60: + log.warning(f"Client was configured to block for up to {start_wait} seconds when initializing. We recommend blocking no longer than 60.") + + if start_wait > 0: + log.info("Waiting up to " + str(start_wait) + " seconds for LaunchDarkly client to initialize...") + await update_processor_ready.wait(start_wait) + + if self.is_initialized() is True: + log.info("Started LaunchDarkly Client: OK") + else: + log.warning("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. " "Feature Flags may not yet be available.") + + async def _create_http_session(self): + """Create and return the aiohttp session. Called from __start_up.""" + import ssl + + import aiohttp + + ssl_ctx = ssl.create_default_context( + cafile=self._config.http.ca_certs or certifi.where() + ) + if self._config.http.cert_file: + ssl_ctx.load_cert_chain(self._config.http.cert_file) + if self._config.http.disable_ssl_verification: + ssl_ctx.check_hostname = False + ssl_ctx.verify_mode = ssl.CERT_NONE + log.warning("TLS verification disabled") + + connector = aiohttp.TCPConnector(ssl=ssl_ctx, limit_per_host=10) + self._proxy = self._config.http.http_proxy + return aiohttp.ClientSession( + connector=connector, + trust_env=(self._proxy is None), + ) + + def _make_data_system(self) -> AsyncDataSystem: + datasystem_config = self._config.datasystem_config + if datasystem_config is None: + from ldclient.impl.datasystem.async_fdv1 import AsyncFDv1 + + return AsyncFDv1(self._config, self._select_feature_store(), self._session, self._proxy) + + raise NotImplementedError("FDv2 is not yet supported in the async client") + + def _select_feature_store(self) -> AsyncFeatureStore: + """Choose the async feature store for the v1 data system based on the + configured store.""" + feature_store = self._config.feature_store + if feature_store is None: + return AsyncInMemoryFeatureStore() + return feature_store + + async def __register_plugins(self, environment_metadata: EnvironmentMetadata): + for plugin in self._config.plugins: + try: + await plugin.register(self, environment_metadata) + except Exception as e: + log.error("Error registering plugin %s: %s", plugin.metadata.name, e) + + def _set_event_processor(self, config): + if config.offline or not config.send_events: + self._event_processor = AsyncNullEventProcessor() + return None + if not config.event_processor_class: + diagnostic_id = create_diagnostic_id(config) + diagnostic_accumulator = None if config.diagnostic_opt_out else _DiagnosticAccumulator(diagnostic_id) + self._event_processor = DefaultAsyncEventProcessor(config, self._session, diagnostic_accumulator=diagnostic_accumulator) + return diagnostic_accumulator + self._event_processor = config.event_processor_class(config) + return None + + def get_sdk_key(self) -> Optional[str]: + """Returns the configured SDK key.""" + return self._config.sdk_key + + def _send_event(self, event): + self._event_processor.send_event(event) + + def track_migration_op(self, tracker: OpTracker): + """ + Tracks the results of a migrations operation. This event includes + measurements which can be used to enhance the observability of a + migration within the LaunchDarkly UI. + + Customers making use of the :class:`ldclient.MigrationBuilder` should + not need to call this method manually. + + Customers not using the builder should provide this method with the + tracker returned from calling :func:`migration_variation`. + """ + event = tracker.build() + + if isinstance(event, str): + log.error("error generting migration op event %s; no event will be emitted", event) + return + + self._send_event(event) + + def track(self, event_name: str, context: Context, data: Optional[Any] = None, metric_value: Optional[AnyNum] = None): + """Tracks that an application-defined event occurred. + + This method creates a "custom" analytics event containing the specified event name (key) + and context properties. You may attach arbitrary data or a metric value to the event with the + optional ``data`` and ``metric_value`` parameters. + + Note that event delivery is asynchronous, so the event may not actually be sent until later; + see :func:`flush()`. + + :param event_name: the name of the event + :param context: the evaluation context associated with the event + :param data: optional additional data associated with the event + :param metric_value: a numeric value used by the LaunchDarkly experimentation feature in + numeric custom metrics; can be omitted if this event is used by only non-numeric metrics + """ + if not context.valid: + log.warning("Invalid context for track (%s)" % context.error) + else: + self._send_event(self._event_factory_default.new_custom_event(event_name, context, data, metric_value)) + + def identify(self, context: Context): + """Reports details about an evaluation context. + + This method simply creates an analytics event containing the context properties, to + that LaunchDarkly will know about that context if it does not already. + + Evaluating a flag, by calling :func:`variation()` or :func:`variation_detail()`, also + sends the context information to LaunchDarkly (if events are enabled), so you only + need to use :func:`identify()` if you want to identify the context without evaluating a + flag. + + :param context: the context to register + """ + + if not context.valid: + log.warning("Invalid context for identify (%s)" % context.error) + else: + self._send_event(self._event_factory_default.new_identify_event(context)) + + def is_offline(self) -> bool: + """Returns true if the client is in offline mode.""" + return self._config.offline + + def is_initialized(self) -> bool: + """Returns true if the client has successfully connected to LaunchDarkly. + + If this returns false, it means that the client has not yet successfully connected to LaunchDarkly. + It might still be in the process of starting up, or it might be attempting to reconnect after an + unsuccessful attempt, or it might have received an unrecoverable error (such as an invalid SDK key) + and given up. + """ + if self.is_offline() or self._config.use_ldd: + return True + + return self._data_system.data_availability.at_least(DataAvailability.CACHED) + + async def flush(self): + """Flushes all pending analytics events. + + Normally, batches of events are delivered in the background at intervals determined by the + ``flush_interval`` property of :class:`ldclient.config.Config`. Calling ``flush()`` + schedules the next event delivery to be as soon as possible; however, the delivery still + happens asynchronously on a worker thread, so this method will return immediately. + """ + if self._config.offline: + return + # flush() only schedules delivery; it does not await, so there is + # nothing to await here. + self._event_processor.flush() + + async def flush_and_wait(self, timeout: float) -> bool: + """Flushes all pending analytics events and waits for delivery to complete. + + Unlike :meth:`flush`, this waits for the buffered events to be delivered, up to ``timeout`` + seconds. Returns True if delivery completed within the timeout, or False if it timed out. + + :param timeout: the maximum number of seconds to wait for delivery + """ + if self._config.offline: + return True + return await self._event_processor.flush_and_wait(timeout) + + async def variation(self, key: str, context: Context, default: Any) -> Any: + """Calculates the value of a feature flag for a given context. + + :param key: the unique key for the feature flag + :param context: the evaluation context + :param default: the default value of the flag, to be used if the value is not + available from LaunchDarkly + :return: the variation for the given context, or the ``default`` value if the flag cannot be evaluated + """ + + async def evaluate(): + detail, _ = await self._evaluate_internal(key, context, default, self._event_factory_default) + return _EvaluationWithHookResult(evaluation_detail=detail) + + return (await self.__evaluate_with_hooks(key=key, context=context, default_value=default, method="variation", block=evaluate)).evaluation_detail.value + + async def variation_detail(self, key: str, context: Context, default: Any) -> EvaluationDetail: + """Calculates the value of a feature flag for a given context, and returns an object that + describes the way the value was determined. + + The ``reason`` property in the result will also be included in analytics events, if you are + capturing detailed event data for this flag. + + :param key: the unique key for the feature flag + :param context: the evaluation context + :param default: the default value of the flag, to be used if the value is not + available from LaunchDarkly + :return: an :class:`ldclient.evaluation.EvaluationDetail` object that includes the feature + flag value and evaluation reason + """ + + async def evaluate(): + detail, _ = await self._evaluate_internal(key, context, default, self._event_factory_with_reasons) + return _EvaluationWithHookResult(evaluation_detail=detail) + + return (await self.__evaluate_with_hooks(key=key, context=context, default_value=default, method="variation_detail", block=evaluate)).evaluation_detail + + async def migration_variation(self, key: str, context: Context, default_stage: Stage) -> Tuple[Stage, OpTracker]: + """ + This method returns the migration stage of the migration feature flag + for the given evaluation context. + + This method returns the default stage if there is an error or the flag + does not exist. If the default stage is not a valid stage, then a + default stage of :class:`ldclient.migrations.Stage.OFF` will be used + instead. + """ + if not isinstance(default_stage, Stage) or default_stage not in Stage: + log.error(f"default stage {default_stage} is not a valid stage; using 'off' instead") + default_stage = Stage.OFF + + async def evaluate(): + detail, flag = await self._evaluate_internal(key, context, default_stage.value, self._event_factory_default) + + if isinstance(detail.value, str): + stage = Stage.from_str(detail.value) + if stage is not None: + tracker = OpTracker(key, flag, context, detail, default_stage) + return _EvaluationWithHookResult(evaluation_detail=detail, results={'default_stage': stage, 'tracker': tracker}) + + detail = EvaluationDetail(default_stage.value, None, error_reason('WRONG_TYPE')) + tracker = OpTracker(key, flag, context, detail, default_stage) + return _EvaluationWithHookResult(evaluation_detail=detail, results={'default_stage': default_stage, 'tracker': tracker}) + + hook_result = await self.__evaluate_with_hooks(key=key, context=context, default_value=default_stage.value, method="migration_variation", block=evaluate) + return hook_result.results['default_stage'], hook_result.results['tracker'] + + async def _evaluate_internal(self, key: str, context: Context, default: Any, event_factory) -> Tuple[EvaluationDetail, Optional[FeatureFlag]]: + default = self._config.get_default(key, default) + + if self._config.offline: + return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None + + if self._data_system.data_availability != DataAvailability.REFRESHED: + if self._data_system.data_availability == DataAvailability.CACHED: + log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key) + else: + log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key) + reason = error_reason('CLIENT_NOT_READY') + self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason)) + return EvaluationDetail(default, None, reason), None + + if not context.valid: + log.warning("Context was invalid for flag evaluation (%s); returning default value" % context.error) + return EvaluationDetail(default, None, error_reason('USER_NOT_SPECIFIED')), None + + try: + flag = await _get_store_item(self._data_system.store, FEATURES, key) + except Exception as e: + log.error("Unexpected error while retrieving feature flag \"%s\": %s" % (key, repr(e))) + log.debug(traceback.format_exc()) + reason = error_reason('EXCEPTION') + self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason)) + return EvaluationDetail(default, None, reason), None + if not flag: + reason = error_reason('FLAG_NOT_FOUND') + self._send_event(event_factory.new_unknown_flag_event(key, context, default, reason)) + return EvaluationDetail(default, None, reason), None + else: + try: + result = await self._evaluator.evaluate(flag, context, event_factory) + for event in result.events or []: + self._send_event(event) + detail = result.detail + if detail.is_default_value(): + detail = EvaluationDetail(default, None, detail.reason) + self._send_event(event_factory.new_eval_event(flag, context, detail, default)) + return detail, flag + except Exception as e: + log.error("Unexpected error while evaluating feature flag \"%s\": %s" % (key, repr(e))) + log.debug(traceback.format_exc()) + reason = error_reason('EXCEPTION') + self._send_event(event_factory.new_default_event(flag, context, default, reason)) + return EvaluationDetail(default, None, reason), flag + + async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState: + """Returns an object that encapsulates the state of all feature flags for a given context, + including the flag values and also metadata that can be used on the front end. See the + JavaScript SDK Reference Guide on + `Bootstrapping `_. + + This method does not send analytics events back to LaunchDarkly. + + :param context: the end context requesting the feature flags + :param kwargs: optional parameters affecting how the state is computed - see below + + :Keyword Arguments: + * **client_side_only** (*boolean*) -- + set to True to limit it to only flags that are marked for use with the client-side SDK + (by default, all flags are included) + * **with_reasons** (*boolean*) -- + set to True to include evaluation reasons in the state (see :func:`variation_detail()`) + * **details_only_for_tracked_flags** (*boolean*) -- + set to True to omit any metadata that is normally only used for event generation, such + as flag versions and evaluation reasons, unless the flag has event tracking or debugging + turned on + + :return: a FeatureFlagsState object (will never be None; its ``valid`` property will be False + if the client is offline, has not been initialized, or the context is invalid) + """ + if self._config.offline: + log.warning("all_flags_state() called, but client is in offline mode. Returning empty state") + return FeatureFlagsState(False) + + if self._data_system.data_availability != DataAvailability.REFRESHED: + if self._data_system.data_availability == DataAvailability.CACHED: + log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store") + else: + log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state") + return FeatureFlagsState(False) + + if not context.valid: + log.warning("Context was invalid for all_flags_state (%s); returning default value" % context.error) + return FeatureFlagsState(False) + + state = FeatureFlagsState(True) + client_only = kwargs.get('client_side_only', False) + with_reasons = kwargs.get('with_reasons', False) + details_only_if_tracked = kwargs.get('details_only_for_tracked_flags', False) + try: + flags_map = await self._data_system.store.all(FEATURES) + if flags_map is None: + raise ValueError("feature store error") + except Exception as e: + log.error("Unable to read flags for all_flag_state: %s" % repr(e)) + return FeatureFlagsState(False) + + for key, flag in flags_map.items(): + if client_only and not flag.get('clientSide', False): + continue + try: + result = await self._evaluator.evaluate(flag, context, self._event_factory_default) + detail = result.detail + except Exception as e: + log.error("Error evaluating flag \"%s\" in all_flags_state: %s" % (key, repr(e))) + log.debug(traceback.format_exc()) + reason = {'kind': 'ERROR', 'errorKind': 'EXCEPTION'} + detail = EvaluationDetail(None, None, reason) + + requires_experiment_data = EventFactory.is_experiment(flag, detail.reason) + flag_state = { + 'key': flag['key'], + 'value': detail.value, + 'variation': detail.variation_index, + 'reason': detail.reason, + 'version': flag['version'], + 'prerequisites': result.prerequisites, + 'trackEvents': flag.get('trackEvents', False) or requires_experiment_data, + 'trackReason': requires_experiment_data, + 'debugEventsUntilDate': flag.get('debugEventsUntilDate', None), + } + + state.add_flag(flag_state, with_reasons, details_only_if_tracked) + + return state + + def secure_mode_hash(self, context: Context) -> str: + """Creates a hash string that can be used by the JavaScript SDK to identify a context. + + For more information, see the documentation on + `Secure mode `_. + + :param context: the evaluation context + :return: the hash string + """ + return _secure_mode_hash(self._config, context) + + def add_hook(self, hook: AsyncHook): + """ + Add a hook to the client. In order to register a hook before the client starts, please use the `hooks` property of + `AsyncConfig`. + + Hooks provide entrypoints which allow for observation of SDK functions. + + The async client only accepts :class:`ldclient.hook.AsyncHook` instances; + passing a synchronous :class:`ldclient.hook.Hook` raises ``TypeError``. + + :param hook: + """ + if not isinstance(hook, AsyncHook): + raise TypeError("AsyncLDClient requires an AsyncHook; synchronous Hook instances are not supported") + + with self.__hooks_lock.write(): + self.__hooks.append(hook) + + async def __evaluate_with_hooks(self, key: str, context: Context, default_value: Any, method: str, block: Callable[[], Any]) -> _EvaluationWithHookResult: + """ + # evaluate_with_hook will run the provided block, wrapping it with evaluation hook support. + # + # :param key: + # :param context: + # :param default: + # :param method: + # :param block: + # :return: + """ + hooks = [] # type: List[AsyncHook] + with self.__hooks_lock.read(): + if len(self.__hooks) == 0: + return await block() + + hooks = self.__hooks.copy() + + series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) + hook_data = await self.__execute_before_evaluation(hooks, series_context) + evaluation_result = await block() + await self.__execute_after_evaluation(hooks, series_context, hook_data, evaluation_result.evaluation_detail) + + return evaluation_result + + async def __execute_before_evaluation(self, hooks: List[AsyncHook], series_context: EvaluationSeriesContext) -> List[dict]: + return [await self.__try_execute_stage("beforeEvaluation", hook.metadata.name, lambda: hook.before_evaluation(series_context, {})) for hook in hooks] + + async def __execute_after_evaluation(self, hooks: List[AsyncHook], series_context: EvaluationSeriesContext, hook_data: List[dict], evaluation_detail: EvaluationDetail) -> List[dict]: + return [ + await self.__try_execute_stage("afterEvaluation", hook.metadata.name, lambda: hook.after_evaluation(series_context, data, evaluation_detail)) + for (hook, data) in reversed(list(zip(hooks, hook_data))) + ] + + async def __try_execute_stage(self, method: str, hook_name: str, block: Callable[[], Any]) -> dict: + try: + return await block() + except BaseException as e: + log.error(f"An error occurred in {method} of the hook {hook_name}: #{e}") + return {} + + @property + def big_segment_store_status_provider(self) -> BigSegmentStoreStatusProvider: + """ + Returns an interface for tracking the status of a Big Segment store. + + The :class:`ldclient.interfaces.BigSegmentStoreStatusProvider` has methods for checking + whether the Big Segment store is (as far as the SDK knows) currently operational and + tracking changes in this status. + """ + return self.__big_segment_store_manager.status_provider + + @property + def data_source_status_provider(self) -> DataSourceStatusProvider: + """ + Returns an interface for tracking the status of the data source. + + The data source is the mechanism that the SDK uses to get feature flag configurations, such + as a streaming connection (the default) or poll requests. The + :class:`ldclient.interfaces.DataSourceStatusProvider` has methods for checking whether the + data source is (as far as the SDK knows) currently operational and tracking changes in this + status. + + :return: The data source status provider + """ + return self.__data_source_status_provider + + @property + def data_store_status_provider(self) -> DataStoreStatusProvider: + """ + Returns an interface for tracking the status of a persistent data store. + + The provider has methods for checking whether the data store is (as far + as the SDK knows) currently operational, tracking changes in this + status, and getting cache statistics. These are only relevant for a + persistent data store; if you are using an in-memory data store, then + this method will return a stub object that provides no information. + + :return: The data store status provider + """ + return self.__data_store_status_provider + + @property + def flag_tracker(self) -> AsyncFlagTracker: + """ + Returns an interface for tracking changes in feature flag configurations. + + The :class:`ldclient.interfaces.AsyncFlagTracker` contains methods for + requesting notifications about feature flag changes using an event + listener model. + """ + if not self._started: + raise RuntimeError("AsyncLDClient.flag_tracker is not available until after start()") + return self.__flag_tracker + + async def __aenter__(self): + await self.start() + return self + + async def __aexit__(self, *args): + await self.close() + + +__all__ = ['AsyncLDClient'] diff --git a/ldclient/client.py b/ldclient/client.py index 28de0eaf..11016572 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -238,7 +238,7 @@ def postfork(self, start_wait: float = 5): def __start_up(self, start_wait: float): environment_metadata = get_environment_metadata(self._config, "python-server-sdk") - plugin_hooks = get_plugin_hooks(self._config, environment_metadata) + plugin_hooks = get_plugin_hooks(self._config.plugins, environment_metadata) self.__hooks_lock = ReadWriteLock() self.__hooks = self._config.hooks + plugin_hooks # type: List[Hook] diff --git a/ldclient/impl/client_common.py b/ldclient/impl/client_common.py index 4cb5d026..a99b5eba 100644 --- a/ldclient/impl/client_common.py +++ b/ldclient/impl/client_common.py @@ -10,15 +10,17 @@ import hashlib import hmac -from typing import List +from typing import List, Sequence, Union -from ldclient.config import Config, SdkIdentityConfig +from ldclient.config import SdkIdentityConfig from ldclient.context import Context -from ldclient.hook import Hook +from ldclient.hook import AsyncHook, Hook from ldclient.impl.util import log from ldclient.plugin import ( ApplicationMetadata, + AsyncPlugin, EnvironmentMetadata, + Plugin, SdkMetadata ) from ldclient.version import VERSION @@ -46,9 +48,9 @@ def get_environment_metadata(config: SdkIdentityConfig, sdk_name: str) -> Enviro ) -def get_plugin_hooks(config: Config, environment_metadata: EnvironmentMetadata) -> List[Hook]: - hooks = [] - for plugin in config.plugins: +def get_plugin_hooks(plugins: Sequence[Union[Plugin, AsyncPlugin]], environment_metadata: EnvironmentMetadata) -> List: + hooks: List = [] + for plugin in plugins: try: hooks.extend(plugin.get_hooks(environment_metadata)) except Exception as e: diff --git a/ldclient/impl/datasystem/__init__.py b/ldclient/impl/datasystem/__init__.py index c1d65a90..d04bdd9b 100644 --- a/ldclient/impl/datasystem/__init__.py +++ b/ldclient/impl/datasystem/__init__.py @@ -6,10 +6,14 @@ from abc import abstractmethod from enum import Enum from threading import Event -from typing import Protocol, runtime_checkable +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +if TYPE_CHECKING: + from ldclient.impl.aio.concurrency import AsyncEvent from ldclient.impl.listeners import Listeners from ldclient.interfaces import ( + AsyncReadOnlyStore, DataSourceStatusProvider, DataStoreStatusProvider, FlagTracker, @@ -143,6 +147,80 @@ def store(self) -> ReadOnlyStore: raise NotImplementedError +class AsyncDataSystem(Protocol): + """ + Async counterpart of :class:`DataSystem`: the same requirements, with the + data system's background work running as asyncio tasks. + """ + + @abstractmethod + def start(self, set_on_ready: "AsyncEvent"): + """ + Starts the data system. + + This method will return immediately. The provided event will be set when the system + has reached an initial state (either permanently failed, e.g. due to bad auth, or + succeeded) + """ + raise NotImplementedError + + @abstractmethod + async def stop(self): + """ + Halts the data system. Should be called when the client is closed to stop any long running + operations. + """ + raise NotImplementedError + + @property + @abstractmethod + def data_source_status_provider(self) -> DataSourceStatusProvider: + """ + Returns an interface for tracking the status of the data source. + """ + raise NotImplementedError + + @property + @abstractmethod + def data_store_status_provider(self) -> DataStoreStatusProvider: + """ + Returns an interface for tracking the status of a persistent data store. + """ + raise NotImplementedError + + @property + @abstractmethod + def flag_change_listeners(self) -> Listeners: + """ + Returns the collection of listeners for flag change events. + """ + raise NotImplementedError + + @property + @abstractmethod + def data_availability(self) -> DataAvailability: + """ + Indicates what form of data is currently available. + """ + raise NotImplementedError + + @property + @abstractmethod + def target_availability(self) -> DataAvailability: + """ + Indicates the ideal form of data attainable given the current configuration. + """ + raise NotImplementedError + + @property + @abstractmethod + def store(self) -> AsyncReadOnlyStore: + """ + Returns the data store used by the data system. + """ + raise NotImplementedError + + class DiagnosticAccumulator(Protocol): def record_stream_init(self, timestamp, duration, failed): raise NotImplementedError diff --git a/ldclient/impl/datasystem/async_fdv1.py b/ldclient/impl/datasystem/async_fdv1.py new file mode 100644 index 00000000..41366257 --- /dev/null +++ b/ldclient/impl/datasystem/async_fdv1.py @@ -0,0 +1,171 @@ +from typing import Any, Optional + +from ldclient.async_config import AsyncConfig +from ldclient.impl.aio.concurrency import AsyncEvent +from ldclient.impl.aio.transport import AsyncHTTPTransport, AsyncSSEFactory +from ldclient.impl.datasource.async_feature_requester import ( + AsyncFeatureRequesterImpl +) +from ldclient.impl.datasource.async_polling import AsyncPollingUpdateProcessor +from ldclient.impl.datasource.async_status import AsyncDataSourceUpdateSinkImpl +from ldclient.impl.datasource.async_streaming import ( + AsyncStreamingUpdateProcessor +) +from ldclient.impl.datasource.status import DataSourceStatusProviderImpl +from ldclient.impl.datastore.status import ( + DataStoreStatusProviderImpl, + DataStoreUpdateSinkImpl +) +from ldclient.impl.datasystem import ( + AsyncDataSystem, + DataAvailability, + DiagnosticAccumulator +) +from ldclient.impl.listeners import Listeners +from ldclient.impl.stubs import AsyncNullUpdateProcessor +from ldclient.impl.util import log +from ldclient.interfaces import ( + AsyncFeatureStore, + AsyncReadOnlyStore, + AsyncUpdateProcessor, + DataSourceStatusProvider, + DataStoreStatusProvider +) + + +class AsyncFDv1(AsyncDataSystem): + """ + AsyncFDv1 provides the v1 data source and store behavior through the + AsyncDataSystem interface. It is the async version of + :class:`ldclient.impl.datasystem.fdv1.FDv1`. Unlike the sync side, it uses + the feature store directly and does not wrap it for persistent-store status + monitoring. + """ + + def __init__(self, config: AsyncConfig, store: AsyncFeatureStore, session: Optional[Any] = None, proxy: Optional[str] = None): + self._config = config + self._store = store + self._session = session + self._proxy = proxy + + # Set up data store status tracking (no store wrapper) + self._data_store_listeners = Listeners() + self._data_store_update_sink = DataStoreUpdateSinkImpl( + self._data_store_listeners + ) + # The provider only calls the store's monitoring methods, which the async + # store also has, so the sync-typed signature is fine. + self._data_store_status_provider_impl = DataStoreStatusProviderImpl( + self._store, self._data_store_update_sink # type: ignore[arg-type] + ) + + # Set up the data source status tracking and listeners + self._data_source_listeners = Listeners() + self._flag_change_listeners = Listeners() + self._data_source_update_sink = AsyncDataSourceUpdateSinkImpl( + self._store, + self._data_source_listeners, + self._flag_change_listeners, + ) + self._data_source_status_provider_impl = DataSourceStatusProviderImpl( + self._data_source_listeners, self._data_source_update_sink + ) + + # v1 processors read the sink from the config for status updates. The config + # attribute is typed as the sync sink, but the async sink has the same methods. + self._config._data_source_update_sink = self._data_source_update_sink # type: ignore[assignment] + + # Update processor created in start(), because it needs the ready event + self._update_processor: Optional[AsyncUpdateProcessor] = None + + # Diagnostic accumulator provided by client for streaming metrics + self._diagnostic_accumulator: Optional[DiagnosticAccumulator] = None + + def start(self, set_on_ready: AsyncEvent): + """ + Starts the v1 update processor and returns immediately. The provided + event is set by the processor upon first successful initialization or + upon permanent failure. + """ + update_processor = self._make_update_processor( + self._config, self._store, set_on_ready + ) + self._update_processor = update_processor + update_processor.start() + + async def stop(self): + if self._update_processor is not None: + await self._update_processor.stop() + + @property + def store(self) -> AsyncReadOnlyStore: + return self._store + + def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): + """ + Sets the diagnostic accumulator for streaming initialization metrics. + This should be called before start() to ensure metrics are collected. + """ + self._diagnostic_accumulator = diagnostic_accumulator + + @property + def data_source_status_provider(self) -> DataSourceStatusProvider: + return self._data_source_status_provider_impl + + @property + def data_store_status_provider(self) -> DataStoreStatusProvider: + return self._data_store_status_provider_impl + + @property + def flag_change_listeners(self) -> Listeners: + return self._flag_change_listeners + + @property + def data_availability(self) -> DataAvailability: + if self._config.offline: + return DataAvailability.DEFAULTS + + if self._update_processor is not None and self._update_processor.initialized(): + return DataAvailability.REFRESHED + + if self._store.initialized: + return DataAvailability.CACHED + + return DataAvailability.DEFAULTS + + @property + def target_availability(self) -> DataAvailability: + if self._config.offline: + return DataAvailability.DEFAULTS + # In LDD mode or normal connected modes, the ideal is to be refreshed + return DataAvailability.REFRESHED + + def _make_update_processor(self, config: AsyncConfig, store: AsyncFeatureStore, ready: AsyncEvent): + # Mirrors FDv1._make_update_processor but builds the async processors + if config.update_processor_class: + log.info("Using user-specified update processor: " + str(config.update_processor_class)) + return config.update_processor_class(config, store, ready) + + if config.offline or config.use_ldd: + return AsyncNullUpdateProcessor(config, store, ready) + + if config.stream: + return AsyncStreamingUpdateProcessor( + config, + store, + ready, + self._diagnostic_accumulator, + AsyncSSEFactory(config, session=self._session, proxy=self._proxy), + ) + + log.info("Disabling streaming API") + log.warning("You should only disable the streaming API if instructed to do so by LaunchDarkly support") + + if config.feature_requester_class: + feature_requester = config.feature_requester_class(config) + else: + feature_requester = AsyncFeatureRequesterImpl( + config, + AsyncHTTPTransport(config, client=self._session), + ) + return AsyncPollingUpdateProcessor(config, feature_requester, store, ready) diff --git a/ldclient/impl/stubs.py b/ldclient/impl/stubs.py index 9743ccfe..897bbc3d 100644 --- a/ldclient/impl/stubs.py +++ b/ldclient/impl/stubs.py @@ -1,4 +1,9 @@ -from ldclient.interfaces import EventProcessor, UpdateProcessor +from ldclient.interfaces import ( + AsyncEventProcessor, + AsyncUpdateProcessor, + EventProcessor, + UpdateProcessor +) class NullEventProcessor(EventProcessor): @@ -36,3 +41,45 @@ def is_alive(self): def initialized(self): return True + + +class AsyncNullEventProcessor(AsyncEventProcessor): + """Async no-op event processor. The async equivalent of + :class:`NullEventProcessor`, so the async client can await ``stop()`` + uniformly whether events are enabled or not.""" + + def start(self): + pass + + async def stop(self): + pass + + def is_alive(self): + return False + + def send_event(self, event): + pass + + def flush(self): + pass + + async def flush_and_wait(self, timeout: float) -> bool: + return True + + +class AsyncNullUpdateProcessor(AsyncUpdateProcessor): + """Async no-op update processor. The async equivalent of + :class:`NullUpdateProcessor`, used by async FDv1 when offline or in LDD + mode so the data system can await ``stop()`` uniformly.""" + + def __init__(self, config, store, ready): + self._ready = ready + + def start(self): + self._ready.set() + + async def stop(self): + pass + + def initialized(self): + return True diff --git a/ldclient/testing/mock_async_components.py b/ldclient/testing/mock_async_components.py index 45e6887d..3d4bbbd3 100644 --- a/ldclient/testing/mock_async_components.py +++ b/ldclient/testing/mock_async_components.py @@ -3,11 +3,11 @@ """ from ldclient.async_feature_store import AsyncInMemoryFeatureStore -from ldclient.interfaces import EventProcessor, UpdateProcessor +from ldclient.interfaces import AsyncEventProcessor, UpdateProcessor -class MockAsyncEventProcessor(EventProcessor): - """A mock EventProcessor that records send_event() calls for testing. +class MockAsyncEventProcessor(AsyncEventProcessor): + """A mock AsyncEventProcessor that records send_event() calls for testing. flush() and stop() are no-ops. """ @@ -21,7 +21,10 @@ def send_event(self, event): def flush(self): pass - def stop(self): + async def flush_and_wait(self, timeout: float) -> bool: + return True + + async def stop(self): pass diff --git a/ldclient/testing/stub_util.py b/ldclient/testing/stub_util.py index d4d45bc0..c546bbe7 100644 --- a/ldclient/testing/stub_util.py +++ b/ldclient/testing/stub_util.py @@ -110,6 +110,11 @@ def status(self): def headers(self): return self._headers + @property + def data(self): + # The HTTPTransport shim reads the body of every response + return b'' + class MockHttp: def __init__(self): diff --git a/ldclient/testing/test_async_client.py b/ldclient/testing/test_async_client.py new file mode 100644 index 00000000..bd1fa59d --- /dev/null +++ b/ldclient/testing/test_async_client.py @@ -0,0 +1,364 @@ +""" +Tests for AsyncLDClient. +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from ldclient.async_client import AsyncLDClient +from ldclient.async_config import AsyncConfig +from ldclient.context import Context +from ldclient.testing.mock_async_components import ( + MockAsyncEventProcessor, + MockAsyncFeatureStore, + MockAsyncUpdateProcessor +) +from ldclient.versioned_data_kind import FEATURES + + +def _offline_config(**kwargs): + """Return an AsyncConfig that uses offline mode to avoid any network connections.""" + return AsyncConfig("test-sdk-key", offline=True, **kwargs) + + +def _make_flag(key: str, value, version: int = 1) -> dict: + """Build a minimal feature flag dict usable with AsyncInMemoryFeatureStore.""" + return { + 'key': key, + 'version': version, + 'on': True, + 'variations': [False, True, value], + 'fallthrough': {'variation': 2}, + 'offVariation': 0, + 'targets': [], + 'rules': [], + 'prerequisites': [], + 'salt': 'abc', + 'deleted': False, + } + + +@pytest.mark.asyncio +async def test_default_variation_returns_default_when_not_started(): + """variation() before start() returns the default value.""" + client = AsyncLDClient(_offline_config()) + context = Context.create('user-1') + result = await client.variation('some-flag', context, 'fallback') + # In offline mode the client behaves as initialized — no network required. + # The flag doesn't exist, so CLIENT_NOT_READY or FLAG_NOT_FOUND — either way default is returned. + assert result == 'fallback' + + +@pytest.mark.asyncio +async def test_variation_returns_flag_value_when_initialized(): + """After start(), variation() returns the stored flag value. + + We use update_processor_class=MockAsyncUpdateProcessor so that the client + considers itself initialized (store.initialized becomes True once the + NullUpdateProcessor fires ready) without any network connection. + """ + store = MockAsyncFeatureStore() + flag = _make_flag('my-flag', 'hello') + await store.force_set(FEATURES, flag) + # Pre-initialize the store so is_initialized() returns True + store._initialized = True + + # Use MockAsyncUpdateProcessor which sets ready immediately + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + client = AsyncLDClient(config) + await client.start(start_wait=1.0) + + context = Context.create('user-1') + result = await client.variation('my-flag', context, 'default') + assert result == 'hello' + + await client.close() + + +@pytest.mark.asyncio +async def test_start_is_idempotent(): + """Calling start() twice does not raise and does not double-initialize.""" + client = AsyncLDClient(_offline_config()) + await client.start() + data_system_after_first = client._data_system + + await client.start() + data_system_after_second = client._data_system + + assert data_system_after_second is data_system_after_first + await client.close() + + +@pytest.mark.asyncio +async def test_close_is_idempotent(): + """Calling close() twice does not raise.""" + client = AsyncLDClient(_offline_config()) + await client.start() + await client.close() + # Second close should be a no-op + await client.close() + + +@pytest.mark.asyncio +async def test_context_manager(): + """async with AsyncLDClient(config) as client: starts and closes the client.""" + async with AsyncLDClient(_offline_config()) as client: + assert client.is_initialized() + # After exiting, closed flag should be set + assert client._closed is True + + +@pytest.mark.asyncio +async def test_flush_delegates_to_event_processor(): + """flush() calls flush() on the underlying event processor.""" + config = AsyncConfig( + "test-sdk-key", + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + client = AsyncLDClient(config) + await client.start() + + # Replace the event processor with a mock that tracks flush calls + mock_ep = MagicMock() + mock_ep.flush = MagicMock(return_value=None) + client._event_processor = mock_ep + + await client.flush() + mock_ep.flush.assert_called_once() + + await client.close() + + +@pytest.mark.asyncio +async def test_flush_is_noop_when_offline(): + """flush() returns without touching the event processor in offline mode.""" + client = AsyncLDClient(_offline_config()) + await client.start() + + mock_ep = MagicMock() + mock_ep.flush = MagicMock(return_value=None) + client._event_processor = mock_ep + + await client.flush() + mock_ep.flush.assert_not_called() + + await client.close() + + +@pytest.mark.asyncio +async def test_migration_variation_returns_default_stage(): + """migration_variation() returns the default stage and a tracker when the flag is missing.""" + from ldclient.migrations import OpTracker, Stage + + async with AsyncLDClient(_offline_config()) as client: + stage, tracker = await client.migration_variation('flag', Context.create('user'), Stage.LIVE) + + assert stage == Stage.LIVE + assert isinstance(tracker, OpTracker) + + +@pytest.mark.asyncio +async def test_hooks_are_invoked_during_variation(): + """Hooks added via add_hook() have before/after called during variation().""" + from ldclient.hook import AsyncHook, Metadata + + class RecordingHook(AsyncHook): + def __init__(self): + self.before_calls = [] + self.after_calls = [] + + @property + def metadata(self): + return Metadata(name='recording-hook') + + async def before_evaluation(self, series_context, data): + self.before_calls.append(series_context) + return data + + async def after_evaluation(self, series_context, data, detail): + self.after_calls.append((series_context, detail)) + return data + + hook = RecordingHook() + # Register the hook via add_hook() after construction. + client = AsyncLDClient(_offline_config()) + client.add_hook(hook) + async with client: + context = Context.create('user-1') + result = await client.variation('some-flag', context, 'default-val') + + assert result == 'default-val' + assert len(hook.before_calls) == 1 + assert hook.before_calls[0].key == 'some-flag' + assert len(hook.after_calls) == 1 + assert hook.after_calls[0][0].key == 'some-flag' + + +@pytest.mark.asyncio +async def test_add_hook_rejects_sync_hook(): + """add_hook() raises TypeError when given a synchronous Hook.""" + from ldclient.hook import EvaluationSeriesContext, Hook, Metadata + + class SyncHook(Hook): + @property + def metadata(self): + return Metadata(name='sync-hook') + + def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: + return data + + def after_evaluation(self, series_context, data, detail): + return data + + client = AsyncLDClient(_offline_config()) + with pytest.raises(TypeError): + client.add_hook(SyncHook()) + + +@pytest.mark.asyncio +async def test_flag_tracker_before_start_raises(): + """Accessing flag_tracker before start() raises RuntimeError.""" + client = AsyncLDClient(_offline_config()) + with pytest.raises(RuntimeError): + _ = client.flag_tracker + + +@pytest.mark.asyncio +async def test_variation_detail_returns_reason(): + """variation_detail() returns an EvaluationDetail with a non-None reason.""" + store = MockAsyncFeatureStore() + flag = _make_flag('detail-flag', 'hello') + await store.force_set(FEATURES, flag) + store._initialized = True + + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + async with AsyncLDClient(config) as client: + context = Context.create('user-1') + detail = await client.variation_detail('detail-flag', context, 'fallback') + + from ldclient.evaluation import EvaluationDetail + assert isinstance(detail, EvaluationDetail) + assert detail.reason is not None + + +@pytest.mark.asyncio +async def test_track_sends_event(): + """track() sends a custom event to the event processor.""" + store = MockAsyncFeatureStore() + store._initialized = True + + mock_ep = MockAsyncEventProcessor() + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + event_processor_class=lambda _cfg: mock_ep, + send_events=True, + ) + async with AsyncLDClient(config) as client: + context = Context.create('user-1') + client.track('my-event', context, {'data': 1}, 3.14) + + assert len(mock_ep.events) == 1 + event = mock_ep.events[0] + # Events are EventInputCustom objects with .key attribute + from ldclient.impl.events.types import EventInputCustom + assert isinstance(event, EventInputCustom) + assert event.key == 'my-event' + + +@pytest.mark.asyncio +async def test_data_source_status_provider_accessible(): + """data_source_status_provider is not None after start().""" + store = MockAsyncFeatureStore() + store._initialized = True + + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + async with AsyncLDClient(config) as client: + assert client.data_source_status_provider is not None + + +@pytest.mark.asyncio +async def test_is_offline_reflects_config(): + """is_offline() returns True when Config is created with offline=True.""" + async with AsyncLDClient(_offline_config()) as client: + assert client.is_offline() is True + + config = AsyncConfig( + "test-sdk-key", + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + async with AsyncLDClient(config) as client: + assert client.is_offline() is False + + +@pytest.mark.asyncio +async def test_hooks_data_isolation(): + """Each hook's before_evaluation receives its own isolated {} — not data from a prior hook.""" + from ldclient.hook import AsyncHook, Metadata + + received_data_by_hook = {} + + class IsolationHook(AsyncHook): + def __init__(self, name, inject_key=None, inject_val=None): + self._name = name + self._inject_key = inject_key + self._inject_val = inject_val + + @property + def metadata(self): + return Metadata(name=self._name) + + async def before_evaluation(self, series_context, data): + # Record a copy of what we received + received_data_by_hook[self._name] = dict(data) + if self._inject_key: + data[self._inject_key] = self._inject_val + return data + + async def after_evaluation(self, series_context, data, detail): + return data + + hook_a = IsolationHook('hook-a', inject_key='hook_a', inject_val=True) + hook_b = IsolationHook('hook-b') + + client = AsyncLDClient(_offline_config()) + client.add_hook(hook_a) + client.add_hook(hook_b) + async with client: + context = Context.create('user-1') + await client.variation('some-flag', context, 'default-val') + + # hook_a received an empty dict + assert received_data_by_hook['hook-a'] == {} + # hook_b also received an empty dict — not hook_a's mutated dict + assert received_data_by_hook['hook-b'] == {} + + +@pytest.mark.asyncio +async def test_start_after_close_raises(): + """Calling start() after close() raises RuntimeError.""" + client = AsyncLDClient(_offline_config()) + await client.start() + await client.close() + with pytest.raises(RuntimeError): + await client.start() diff --git a/ldclient/testing/test_sync_async_parity.py b/ldclient/testing/test_sync_async_parity.py new file mode 100644 index 00000000..72faf6a4 --- /dev/null +++ b/ldclient/testing/test_sync_async_parity.py @@ -0,0 +1,73 @@ +""" +Sync/async public-API parity guard. + +The SDK hand-maintains parallel sync (``foo.py``) and async (``async_foo.py``) +implementations. This test catches the most likely drift -- a public method or +property added, removed, or renamed on one side and forgotten on the other -- +without flagging the legitimate body differences between siblings (async/await, +asyncio.gather vs ThreadPoolExecutor, reworded docstrings). Behavioral drift +inside a shared method body is the job of the contract suites, not a source diff. +""" +import inspect + +import pytest + +from ldclient.async_client import AsyncLDClient +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.client import LDClient +from ldclient.feature_store import InMemoryFeatureStore +from ldclient.impl.async_evaluator import AsyncEvaluator +from ldclient.impl.evaluator import Evaluator +from ldclient.migrations import AsyncMigratorBuilder, MigratorBuilder + + +def _public_surface(cls) -> set: + """All public names on the class -- methods AND properties (names not + starting with ``_``, excluding the ``object`` baseline).""" + return {n for n in dir(cls) if not n.startswith("_")} - set(dir(object)) + + +# (sync_cls, async_cls, sync_only, async_only) +# The allowlists document intentionally one-sided public members. Keep them +# small and justified -- every entry is a place the two APIs deliberately differ. +PAIRS = [ + pytest.param( + LDClient, AsyncLDClient, + {"postfork"}, # sync-only: os.fork() recovery hook (no async equivalent) + # async-only: explicit `await start()` lifecycle, and flush_and_wait + # (the sync client only offers fire-and-forget flush()). + {"start", "flush_and_wait"}, + id="client", + ), + pytest.param( + InMemoryFeatureStore, AsyncInMemoryFeatureStore, + set(), + {"close"}, # async-only: the async FeatureStore interface declares + # `async def close()` for resource teardown; the sync + # FeatureStore has no close() (it is hasattr-guarded at + # every call site). + id="feature_store", + ), + pytest.param(Evaluator, AsyncEvaluator, set(), set(), id="evaluator"), + pytest.param(MigratorBuilder, AsyncMigratorBuilder, set(), set(), id="migrator_builder"), +] + + +@pytest.mark.parametrize("sync_cls, async_cls, sync_only, async_only", PAIRS) +def test_public_surface_parity(sync_cls, async_cls, sync_only, async_only): + sync_surface = _public_surface(sync_cls) - sync_only + async_surface = _public_surface(async_cls) - async_only + + missing_on_async = sync_surface - async_surface + missing_on_sync = async_surface - sync_surface + + assert not missing_on_async, ( + f"{async_cls.__name__} is missing public members present on " + f"{sync_cls.__name__}: {sorted(missing_on_async)} -- add them to the async " + f"sibling, or add to the allowlist if intentionally one-sided." + ) + assert not missing_on_sync, ( + f"{sync_cls.__name__} is missing public members present on " + f"{async_cls.__name__}: {sorted(missing_on_sync)} -- add them to the sync " + f"sibling, or add to the allowlist if intentionally one-sided." + ) From e1acdf66ee2887bcaee926e583c2f322d1dac790 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 5 Aug 2026 17:17:43 -0600 Subject: [PATCH 2/4] fix: Snapshot hooks before awaiting to avoid an event-loop deadlock Address review findings in the async client evaluation and shutdown paths: - __evaluate_with_hooks held the hooks read lock across `await block()` on the empty-hooks fast path. A concurrent sync add_hook() takes the write lock with a blocking wait, freezing the event-loop thread. Snapshot the hooks under the lock and release it before awaiting, mirroring the sync client's no-await safety. - all_flags_state referenced `result.prerequisites` unconditionally even when a per-flag evaluation raised, causing UnboundLocalError on the first failure or reuse of a neighbor's prerequisites on a later one. Bind the prerequisites safely in both branches so an error degrades only that flag. - __try_execute_stage swallowed asyncio.CancelledError via `except BaseException`, defeating cancellation and shutdown. Re-raise it. - _close_components stopped components in sequence with no isolation, so one failing stop() skipped the rest. Stop each component in its own try/except. --- ldclient/async_client.py | 39 +++++++++++++++++++++++++++++++-------- 1 file changed, 31 insertions(+), 8 deletions(-) diff --git a/ldclient/async_client.py b/ldclient/async_client.py index 753441df..d7308d66 100644 --- a/ldclient/async_client.py +++ b/ldclient/async_client.py @@ -197,9 +197,20 @@ async def _close_components(self): """Releases the threads and network connections used by the SDK components. The public :meth:`close` wraps this with a timeout.""" log.info("Closing LaunchDarkly client..") - await self._event_processor.stop() - await self._data_system.stop() - await self.__big_segment_store_manager.stop() + # Stop each component in isolation so one failure does not prevent the + # others from stopping. + try: + await self._event_processor.stop() + except Exception as e: + log.warning("Error stopping event processor: %s", e) + try: + await self._data_system.stop() + except Exception as e: + log.warning("Error stopping data system: %s", e) + try: + await self.__big_segment_store_manager.stop() + except Exception as e: + log.warning("Error stopping big segment store manager: %s", e) async def __start_up(self, start_wait: float): environment_metadata = get_environment_metadata(self._config, "python-server-sdk-async") @@ -611,6 +622,7 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState for key, flag in flags_map.items(): if client_only and not flag.get('clientSide', False): continue + result = None try: result = await self._evaluator.evaluate(flag, context, self._event_factory_default) detail = result.detail @@ -620,6 +632,10 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState reason = {'kind': 'ERROR', 'errorKind': 'EXCEPTION'} detail = EvaluationDetail(None, None, reason) + # A per-flag error leaves result unset; degrade only that flag + # rather than aborting the whole payload or reusing a neighbor's + # prerequisites. + prerequisites = result.prerequisites if result is not None else [] requires_experiment_data = EventFactory.is_experiment(flag, detail.reason) flag_state = { 'key': flag['key'], @@ -627,7 +643,7 @@ async def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState 'variation': detail.variation_index, 'reason': detail.reason, 'version': flag['version'], - 'prerequisites': result.prerequisites, + 'prerequisites': prerequisites, 'trackEvents': flag.get('trackEvents', False) or requires_experiment_data, 'trackReason': requires_experiment_data, 'debugEventsUntilDate': flag.get('debugEventsUntilDate', None), @@ -677,12 +693,16 @@ async def __evaluate_with_hooks(self, key: str, context: Context, default_value: # :param block: # :return: """ - hooks = [] # type: List[AsyncHook] + # Snapshot the hooks under the lock and release it before awaiting. + # A concurrent sync add_hook() takes the write lock with a blocking + # wait; holding the read lock across an await would block the event + # loop and deadlock. See the sync client, which is safe because it + # returns without awaiting. with self.__hooks_lock.read(): - if len(self.__hooks) == 0: - return await block() + hooks = self.__hooks.copy() # type: List[AsyncHook] - hooks = self.__hooks.copy() + if not hooks: + return await block() series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) hook_data = await self.__execute_before_evaluation(hooks, series_context) @@ -703,6 +723,9 @@ async def __execute_after_evaluation(self, hooks: List[AsyncHook], series_contex async def __try_execute_stage(self, method: str, hook_name: str, block: Callable[[], Any]) -> dict: try: return await block() + except asyncio.CancelledError: + # Do not swallow cancellation; it must propagate for shutdown. + raise except BaseException as e: log.error(f"An error occurred in {method} of the hook {hook_name}: #{e}") return {} From 0e65a656d3c06aabb4ba796dd1ef76d4d27c8ccb Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 5 Aug 2026 17:17:50 -0600 Subject: [PATCH 3/4] fix: Keep AsyncLDClient out of __all__ so star-import avoids aiohttp --- ldclient/__init__.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/ldclient/__init__.py b/ldclient/__init__.py index 978d488a..b6b6bf43 100644 --- a/ldclient/__init__.py +++ b/ldclient/__init__.py @@ -95,4 +95,7 @@ def __getattr__(name): raise AttributeError("module 'ldclient' has no attribute %r" % name) -__all__ = ['AsyncLDClient', 'Config', 'Context', 'ContextBuilder', 'ContextMultiBuilder', 'LDClient', 'Result', 'client', 'context', 'evaluation', 'integrations', 'interfaces', 'migrations'] +# AsyncLDClient is intentionally omitted from __all__ so that +# `from ldclient import *` does not import aiohttp. It still resolves lazily +# via __getattr__ as `ldclient.AsyncLDClient`. +__all__ = ['Config', 'Context', 'ContextBuilder', 'ContextMultiBuilder', 'LDClient', 'Result', 'client', 'context', 'evaluation', 'integrations', 'interfaces', 'migrations'] From c0daded2f211cba27667402641850c20e8b6d14a Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Wed, 5 Aug 2026 17:17:50 -0600 Subject: [PATCH 4/4] test: Cover all_flags_state normal and evaluator-raises paths --- ldclient/testing/test_async_client.py | 76 +++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/ldclient/testing/test_async_client.py b/ldclient/testing/test_async_client.py index bd1fa59d..c62ca8bd 100644 --- a/ldclient/testing/test_async_client.py +++ b/ldclient/testing/test_async_client.py @@ -362,3 +362,79 @@ async def test_start_after_close_raises(): await client.close() with pytest.raises(RuntimeError): await client.start() + + +@pytest.mark.asyncio +async def test_all_flags_state_returns_flag_values(): + """all_flags_state() returns a valid state with each flag's value.""" + store = MockAsyncFeatureStore() + # init() decodes the flag dicts into model objects, matching the real + # data-source flow that all_flags_state() reads back. + await store.init({FEATURES: { + 'flag-a': _make_flag('flag-a', 'value-a'), + 'flag-b': _make_flag('flag-b', 'value-b'), + }}) + + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + async with AsyncLDClient(config) as client: + context = Context.create('user-1') + state = await client.all_flags_state(context) + + assert state.valid + assert state.to_values_map() == {'flag-a': 'value-a', 'flag-b': 'value-b'} + + +@pytest.mark.asyncio +async def test_all_flags_state_degrades_gracefully_when_evaluator_raises(): + """A per-flag evaluation error degrades only that flag: the payload is not + aborted, and the failed flag does not reuse a neighbor's prerequisites.""" + from ldclient.evaluation import EvaluationDetail + from ldclient.impl.evaluator_common import EvalResult + + store = MockAsyncFeatureStore() + # Insertion order is preserved: bad-first tests that a first-flag failure + # does not raise UnboundLocalError; good-then-bad tests that the trailing + # failed flag does not inherit the good flag's prerequisites. + await store.init({FEATURES: { + 'flag-bad-first': _make_flag('flag-bad-first', 'x'), + 'flag-good': _make_flag('flag-good', 'value-good'), + 'flag-bad-last': _make_flag('flag-bad-last', 'y'), + }}) + + config = AsyncConfig( + "test-sdk-key", + feature_store=store, + update_processor_class=MockAsyncUpdateProcessor, + send_events=False, + ) + async with AsyncLDClient(config) as client: + async def fake_evaluate(flag, context, event_factory): + if flag['key'].startswith('flag-bad'): + raise RuntimeError("boom") + result = EvalResult() + result.detail = EvaluationDetail('value-good', 2, {'kind': 'FALLTHROUGH'}) + result.prerequisites = ['prereq-x'] + return result + + client._evaluator.evaluate = fake_evaluate + context = Context.create('user-1') + state = await client.all_flags_state(context, with_reasons=True) + + # The payload was not aborted by the first flag raising. + assert state.valid + # The good flag still evaluated normally. + assert state.get_flag_value('flag-good') == 'value-good' + # The failed flags degraded to None rather than raising. + assert state.get_flag_value('flag-bad-first') is None + assert state.get_flag_value('flag-bad-last') is None + + # The trailing failed flag did not inherit the good flag's prerequisites. + flags_state = state.to_json_dict()['$flagsState'] + assert flags_state['flag-good'].get('prerequisites') == ['prereq-x'] + assert 'prerequisites' not in flags_state['flag-bad-last'] + assert 'prerequisites' not in flags_state['flag-bad-first']