diff --git a/ldclient/async_feature_store.py b/ldclient/async_feature_store.py new file mode 100644 index 00000000..5c341fa7 --- /dev/null +++ b/ldclient/async_feature_store.py @@ -0,0 +1,98 @@ +""" +This submodule contains the async in-memory feature store implementation. +""" + +from collections import defaultdict +from typing import Any, Dict, Mapping, Optional + +from ldclient.impl.util import log +from ldclient.interfaces import AsyncFeatureStore, DiagnosticDescription +from ldclient.versioned_data_kind import VersionedDataKind + + +class AsyncInMemoryFeatureStore(AsyncFeatureStore, DiagnosticDescription): + """The default async feature store implementation, which holds all data in memory. + + .. 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. + """ + + def __init__(self): + """Constructs an instance of AsyncInMemoryFeatureStore.""" + # No lock is needed: every method below reads and mutates ``_items`` + # without awaiting in between, so it runs to completion without the + # event loop switching to another coroutine. (The sync twin uses a + # real ``ReadWriteLock`` because threads can preempt each other.) + self._initialized = False + self._items: Dict[VersionedDataKind, Dict[str, Any]] = defaultdict(dict) + + def is_monitoring_enabled(self) -> bool: + return False + + def is_available(self) -> bool: + return True + + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + """ """ + items_of_kind = self._items[kind] + item = items_of_kind.get(key) + if item is None: + log.debug("Attempted to get missing key %s in '%s', returning None", key, kind.namespace) + return None + if 'deleted' in item and item['deleted']: + log.debug("Attempted to get deleted key %s in '%s', returning None", key, kind.namespace) + return None + return item + + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + """ """ + items_of_kind = self._items[kind] + return dict((k, i) for k, i in items_of_kind.items() if ('deleted' not in i) or not i['deleted']) + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + """ """ + all_decoded = {} + for kind, items in all_data.items(): + items_decoded = {} + for key, item in items.items(): + items_decoded[key] = kind.decode(item) + all_decoded[kind] = items_decoded + self._items.clear() + self._items.update(all_decoded) + self._initialized = True + for k in all_data: + log.debug("Initialized '%s' store with %d items", k.namespace, len(all_data[k])) + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> None: + """ """ + items_of_kind = self._items[kind] + i = items_of_kind.get(key) + if i is None or i['version'] < version: + items_of_kind[key] = {'deleted': True, 'version': version} + + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + """ """ + decoded_item = kind.decode(item) + key = item['key'] + items_of_kind = self._items[kind] + i = items_of_kind.get(key) + if i is None or i['version'] < item['version']: + items_of_kind[key] = decoded_item + log.debug("Updated %s in '%s' to version %d", key, kind.namespace, item['version']) + return True + return False + + @property + def initialized(self) -> bool: + """ """ + return self._initialized + + async def close(self) -> None: + """ """ + pass + + def describe_configuration(self, config) -> str: + return 'memory' diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index c02e26d6..4d1e6363 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -6,7 +6,17 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum -from typing import Any, Callable, Generator, List, Mapping, Optional, Protocol +from typing import ( + Any, + AsyncGenerator, + Callable, + Dict, + Generator, + List, + Mapping, + Optional, + Protocol +) from ldclient.context import Context from ldclient.impl.listeners import Listeners @@ -294,6 +304,116 @@ def initialized_internal(self) -> bool: # """ +class AsyncReadOnlyStore(Protocol): + """Read-only async view of a data store used by the async client's evaluation path. + + Both async data systems expose their active store through this uniform interface: + :class:`AsyncFeatureStore` satisfies it directly (FDv1), and FDv2 adapts its + synchronous in-memory store to it. This is the async analog of :class:`ReadOnlyStore`. + """ + + @abstractmethod + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + raise NotImplementedError + + @abstractmethod + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + raise NotImplementedError + + +class AsyncFeatureStore(ABC): + """ + Async interface for a versioned store for feature flags and related objects received from + LaunchDarkly. Implementations should be safe for concurrent access within a single asyncio + event loop. + + .. 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. + + An "object", for ``AsyncFeatureStore``, is simply a dict of arbitrary data which must have + at least three properties: ``key`` (its unique key), ``version`` (the version number provided + by LaunchDarkly), and ``deleted`` (True if this is a placeholder for a deleted object). + + Delete and upsert requests are versioned: if the version number in the request is less than + the currently stored version of the object, the request should be ignored. + """ + + @abstractmethod + async def get(self, kind: VersionedDataKind, key: str) -> Optional[Any]: + """ + Retrieves the object to which the specified key is mapped, or None if the key is not + found or the associated object has a ``deleted`` property of True. + + :param kind: The kind of object to get + :param key: The key whose associated object is to be returned + :return: The object, or None + """ + + @abstractmethod + async def all(self, kind: VersionedDataKind) -> Dict[str, Any]: + """ + Retrieves a dictionary of all associated objects of a given kind, excluding deleted items. + + :param kind: The kind of objects to get + :return: A dictionary of keys to objects + """ + + @abstractmethod + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + """ + Initializes (or re-initializes) the store with the specified set of objects. Any existing + entries will be removed. Implementations can assume that this set of objects is up to + date-- there is no need to perform individual version comparisons between the existing + objects and the supplied data. + + :param all_data: All objects to be stored + """ + + @abstractmethod + async def upsert(self, kind: VersionedDataKind, item: dict) -> bool: + """ + Updates or inserts the object associated with the specified key. If an item with the same + key already exists, it should update it only if the new item's version property is greater + than the old one. + + :param kind: The kind of object to update + :param item: The object to update or insert + :return: True if the store was updated (the item was new or a newer version); False if the + update was rejected because the stored version was equal or newer + """ + + @abstractmethod + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> None: + """ + Deletes the object associated with the specified key, if it exists and its version is less + than the specified version. The object should be replaced in the data store by a + placeholder with the specified version and a "deleted" property of True. + + :param kind: The kind of object to delete + :param key: The key of the object to be deleted + :param version: The version for the delete operation + """ + + @property + @abstractmethod + def initialized(self) -> bool: + """ + Returns whether the store has been initialized yet or not. + """ + + async def close(self) -> None: + """ + Releases any resources used by the store implementation. + + The default implementation is a no-op. Persistent store implementations should + override this to close connections and release resources on shutdown. + """ + pass + + # Internal use only. Common methods for components that perform a task in the background. class BackgroundOperation: @@ -459,6 +579,49 @@ def stop(self): pass +class AsyncBigSegmentStore(ABC): + """ + Async interface for a read-only data store that allows querying of user membership in Big + Segments. + + .. 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. + + Big Segments are a specific type of user segments. For more information, read the LaunchDarkly + documentation: https://docs.launchdarkly.com/home/users/big-segments + """ + + @abstractmethod + async def get_metadata(self) -> BigSegmentStoreMetadata: + """ + Returns information about the overall state of the store. This method will be called only + when the SDK needs the latest state, so it should not be cached. + + :return: the store metadata + """ + pass + + @abstractmethod + async def get_membership(self, context_hash: str) -> Optional[dict]: + """ + Queries the store for a snapshot of the current segment state for a specific context. + + :param context_hash: the hashed context key + :return: True/False values for Big Segments that reference this context, or None + """ + pass + + @abstractmethod + async def stop(self) -> None: + """ + Shuts down the store component and releases any resources it is using. + """ + pass + + class BigSegmentStoreStatus: """ Information about the state of a Big Segment store, provided by :class:`BigSegmentStoreStatusProvider`. @@ -833,6 +996,69 @@ def update_status(self, new_state: DataSourceState, new_error: Optional[DataSour pass +class AsyncDataSourceUpdateSink(ABC): + """ + Async interface that a data source implementation will use to push data into the SDK. + + .. 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. + + The data source interacts with this object, rather than manipulating the data store directly, + so that the SDK can perform any other necessary operations that must happen when data is + updated. + """ + + @abstractmethod + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + """ + Initializes (or re-initializes) the store with the specified set of entities. Any + existing entries will be removed. + + :param all_data: All objects to be stored + """ + pass + + @abstractmethod + async def upsert(self, kind: VersionedDataKind, item: dict) -> None: + """ + Attempt to add an entity, or update an existing entity with the same key. An update + should only succeed if the new item's version is greater than the old one; + otherwise, the method should do nothing. + + :param kind: The kind of object to update + :param item: The object to update or insert + """ + pass + + @abstractmethod + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> None: + """ + Attempt to delete an entity if it exists. Deletion should only succeed if the + version parameter is greater than the existing entity's version; otherwise, the + method should do nothing. + + :param kind: The kind of object to delete + :param key: The key of the object to be deleted + :param version: The version for the delete operation + """ + pass + + @abstractmethod + def update_status(self, new_state: DataSourceState, new_error: Optional[DataSourceErrorInfo]) -> None: + """ + Informs the SDK of a change in the data source's status. + + This method is synchronous (not async) so it can be called from any context. + + :param new_state: The updated state of the data source + :param new_error: An optional error if the new state is an error condition + """ + pass + + class FlagChange: """ Change event fired when some aspect of the flag referenced by the key has changed. @@ -1593,3 +1819,83 @@ def stop(self): to exit as soon as possible. """ raise NotImplementedError + + +class AsyncInitializer(Protocol): # pylint: disable=too-few-public-methods + """ + AsyncInitializer represents a component capable of retrieving a single data result + asynchronously, such as from the LD polling API. + + .. 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. + + The intent of initializers is to quickly fetch an initial set of data, which may be stale but + is fast to retrieve. This initial data serves as a foundation for a Synchronizer to build upon, + enabling it to provide updates as new changes occur. + """ + + @property + @abstractmethod + def name(self) -> str: + """ + Returns the name of the initializer, which is used for logging and debugging. + """ + raise NotImplementedError + + @abstractmethod + async def fetch(self, ss: SelectorStore) -> BasisResult: + """ + fetch should retrieve the initial data set for the data source asynchronously, returning + a Basis object on success, or an error message on failure. + + :param ss: A SelectorStore that provides the Selector to use as a basis for data retrieval. + """ + raise NotImplementedError + + +class AsyncSynchronizer(Protocol): # pylint: disable=too-few-public-methods + """ + AsyncSynchronizer represents a component capable of synchronizing data from an external + data source asynchronously, such as a streaming or polling API. + + .. 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. + + It is responsible for yielding Update objects that represent the current state of the data + source, including any changes that have occurred since the last synchronization. + """ + + @property + @abstractmethod + def name(self) -> str: + """ + Returns the name of the synchronizer, which is used for logging and debugging. + """ + raise NotImplementedError + + @abstractmethod + def sync(self, ss: SelectorStore) -> AsyncGenerator[Update, None]: + """ + sync should begin the synchronization process for the data source, yielding Update objects + asynchronously until the connection is closed or an unrecoverable error occurs. + + Implementations write this as ``async def sync(): yield ...`` (an async generator) and + callers consume it as ``async for update in impl.sync(ss):``. + + :param ss: A SelectorStore that provides the Selector to use as a basis for data retrieval. + """ + raise NotImplementedError + + @abstractmethod + async def stop(self) -> None: + """ + stop should halt the synchronization process, causing the sync method to exit as soon as + possible. + """ + raise NotImplementedError diff --git a/ldclient/testing/async_feature_store_test_base.py b/ldclient/testing/async_feature_store_test_base.py new file mode 100644 index 00000000..b4d594ca --- /dev/null +++ b/ldclient/testing/async_feature_store_test_base.py @@ -0,0 +1,162 @@ +""" +Async equivalent of feature_store_test_base.py. Provides a reusable test suite that can be +run against any AsyncFeatureStore implementation. +""" +from abc import abstractmethod + +import pytest + +from ldclient.interfaces import AsyncFeatureStore +from ldclient.testing.builders import FlagBuilder +from ldclient.versioned_data_kind import FEATURES + + +class AsyncFeatureStoreTester: + @abstractmethod + async def create_feature_store(self) -> AsyncFeatureStore: + pass + + +class AsyncFeatureStoreTestBase: + """Base class for async feature store tests. + + Subclasses must provide a pytest fixture called ``tester`` that returns an instance of + ``AsyncFeatureStoreTester``. Each test method receives the tester as a parameter and creates + its own store instance to ensure test isolation. + """ + + @abstractmethod + def all_testers(self): + pass + + @staticmethod + def make_feature(key, ver): + return FlagBuilder(key).version(ver).on(True).variations(True, False).salt('abc').build() + + async def make_inited_store(self, tester) -> AsyncFeatureStore: + store = await tester.create_feature_store() + await store.init( + { + FEATURES: { + 'foo': self.make_feature('foo', 10).to_json_dict(), + 'bar': self.make_feature('bar', 10).to_json_dict(), + } + } + ) + return store + + @pytest.mark.asyncio + async def test_not_initialized_before_init(self, tester): + store = await tester.create_feature_store() + assert store.initialized is False + + @pytest.mark.asyncio + async def test_initialized(self, tester): + store = await self.make_inited_store(tester) + assert store.initialized is True + + @pytest.mark.asyncio + async def test_get_existing_feature(self, tester): + store = await self.make_inited_store(tester) + expected = self.make_feature('foo', 10) + flag = await store.get(FEATURES, 'foo') + assert flag == expected + + @pytest.mark.asyncio + async def test_get_nonexisting_feature(self, tester): + store = await self.make_inited_store(tester) + assert await store.get(FEATURES, 'biz') is None + + @pytest.mark.asyncio + async def test_get_all_versions(self, tester): + store = await self.make_inited_store(tester) + result = await store.all(FEATURES) + assert len(result) == 2 + assert result.get('foo') == self.make_feature('foo', 10) + assert result.get('bar') == self.make_feature('bar', 10) + + @pytest.mark.asyncio + async def test_upsert_with_newer_version(self, tester): + store = await self.make_inited_store(tester) + new_ver = self.make_feature('foo', 11) + await store.upsert(FEATURES, new_ver) + assert await store.get(FEATURES, 'foo') == new_ver + + @pytest.mark.asyncio + async def test_upsert_with_older_version(self, tester): + store = await self.make_inited_store(tester) + new_ver = self.make_feature('foo', 9) + expected = self.make_feature('foo', 10) + await store.upsert(FEATURES, new_ver) + assert await store.get(FEATURES, 'foo') == expected + + @pytest.mark.asyncio + async def test_upsert_with_new_feature(self, tester): + store = await self.make_inited_store(tester) + new_ver = self.make_feature('biz', 1) + await store.upsert(FEATURES, new_ver) + assert await store.get(FEATURES, 'biz') == new_ver + + @pytest.mark.asyncio + async def test_delete_with_newer_version(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'foo', 11) + assert await store.get(FEATURES, 'foo') is None + + @pytest.mark.asyncio + async def test_delete_unknown_feature(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'biz', 11) + assert await store.get(FEATURES, 'biz') is None + + @pytest.mark.asyncio + async def test_delete_with_older_version(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'foo', 9) + expected = self.make_feature('foo', 10) + assert await store.get(FEATURES, 'foo') == expected + + @pytest.mark.asyncio + async def test_upsert_older_version_after_delete(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'foo', 11) + old_ver = self.make_feature('foo', 9) + await store.upsert(FEATURES, old_ver) + assert await store.get(FEATURES, 'foo') is None + + @pytest.mark.asyncio + async def test_deleted_item_not_in_all(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'foo', 11) + result = await store.all(FEATURES) + assert 'foo' not in result + assert 'bar' in result + + @pytest.mark.asyncio + async def test_upsert_with_equal_version_not_applied(self, tester): + store = await self.make_inited_store(tester) + same_ver = self.make_feature('foo', 10) + expected = self.make_feature('foo', 10) + await store.upsert(FEATURES, same_ver) + assert await store.get(FEATURES, 'foo') == expected + + @pytest.mark.asyncio + async def test_delete_with_equal_version_not_applied(self, tester): + store = await self.make_inited_store(tester) + await store.delete(FEATURES, 'foo', 10) + expected = self.make_feature('foo', 10) + assert await store.get(FEATURES, 'foo') == expected + + @pytest.mark.asyncio + async def test_init_replaces_existing_data(self, tester): + store = await tester.create_feature_store() + await store.upsert(FEATURES, self.make_feature('foo', 1)) + await store.init( + { + FEATURES: { + 'bar': self.make_feature('bar', 1).to_json_dict(), + } + } + ) + assert await store.get(FEATURES, 'foo') is None + assert await store.get(FEATURES, 'bar') == self.make_feature('bar', 1) diff --git a/ldclient/testing/mock_async_components.py b/ldclient/testing/mock_async_components.py new file mode 100644 index 00000000..45e6887d --- /dev/null +++ b/ldclient/testing/mock_async_components.py @@ -0,0 +1,77 @@ +""" +Test utilities for async SDK components. +""" + +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.interfaces import EventProcessor, UpdateProcessor + + +class MockAsyncEventProcessor(EventProcessor): + """A mock EventProcessor that records send_event() calls for testing. + + flush() and stop() are no-ops. + """ + + def __init__(self): + self.events = [] + + def send_event(self, event): + self.events.append(event) + + def flush(self): + pass + + def stop(self): + pass + + +class MockAsyncFeatureStore(AsyncInMemoryFeatureStore): + """A test wrapper around AsyncInMemoryFeatureStore that adds helper methods for test setup + and records init() calls. + """ + + def __init__(self): + super().__init__() + self.inits = [] + + async def init(self, all_data): + self.inits.append(all_data) + await super().init(all_data) + + async def force_set(self, kind, item): + """Directly insert an item into the store, bypassing version checks. + + Useful for setting up test state without going through normal upsert semantics. + """ + self._items[kind][item['key']] = item + + async def force_delete(self, kind, key): + """Directly remove an item from the store. + + Useful for tearing down test state without going through the delete tombstone mechanism. + """ + self._items[kind].pop(key, None) + + +class MockAsyncUpdateProcessor(UpdateProcessor): + """A mock UpdateProcessor that immediately reports itself as initialized. + + Used to fake a ready data source in client tests. If a ``ready`` event is provided, it is + set immediately in the constructor, matching the behavior of the sync ``MockUpdateProcessor``. + """ + + def __init__(self, config=None, store=None, ready=None): + if ready is not None: + ready.set() + + def start(self): + pass + + def stop(self): + pass + + def initialized(self) -> bool: + return True + + def is_alive(self) -> bool: + return True diff --git a/ldclient/testing/test_async_in_memory_feature_store.py b/ldclient/testing/test_async_in_memory_feature_store.py new file mode 100644 index 00000000..afed356b --- /dev/null +++ b/ldclient/testing/test_async_in_memory_feature_store.py @@ -0,0 +1,26 @@ +import pytest + +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.interfaces import AsyncFeatureStore +from ldclient.testing.async_feature_store_test_base import ( + AsyncFeatureStoreTestBase, + AsyncFeatureStoreTester +) + + +def test_async_in_memory_status_checks(): + store = AsyncInMemoryFeatureStore() + + assert store.is_monitoring_enabled() is False + assert store.is_available() is True + + +class AsyncInMemoryFeatureStoreTester(AsyncFeatureStoreTester): + async def create_feature_store(self) -> AsyncFeatureStore: + return AsyncInMemoryFeatureStore() + + +class TestAsyncInMemoryFeatureStore(AsyncFeatureStoreTestBase): + @pytest.fixture + def tester(self): + return AsyncInMemoryFeatureStoreTester()