diff --git a/openfeature/_event_support.py b/openfeature/_event_support.py new file mode 100644 index 00000000..c15d68b0 --- /dev/null +++ b/openfeature/_event_support.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from collections import defaultdict +from typing import TYPE_CHECKING, Dict, List + +from openfeature.event import ( + EventDetails, + EventHandler, + ProviderEvent, + ProviderEventDetails, +) +from openfeature.provider import FeatureProvider + +if TYPE_CHECKING: + from openfeature.client import OpenFeatureClient + + +class EventSupport: + _global_handlers: Dict[ProviderEvent, List[EventHandler]] + _client_handlers: Dict[OpenFeatureClient, Dict[ProviderEvent, List[EventHandler]]] + + def __init__(self) -> None: + self._global_handlers = defaultdict(list) + self._client_handlers = defaultdict(lambda: defaultdict(list)) + + def run_client_handlers( + self, client: OpenFeatureClient, event: ProviderEvent, details: EventDetails + ) -> None: + for handler in self._client_handlers[client][event]: + handler(details) + + def run_global_handlers(self, event: ProviderEvent, details: EventDetails) -> None: + for handler in self._global_handlers[event]: + handler(details) + + def add_client_handler( + self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler + ) -> None: + handlers = self._client_handlers[client][event] + handlers.append(handler) + + self._run_immediate_handler(client, event, handler) + + def remove_client_handler( + self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler + ) -> None: + handlers = self._client_handlers[client][event] + handlers.remove(handler) + + def add_global_handler(self, event: ProviderEvent, handler: EventHandler) -> None: + self._global_handlers[event].append(handler) + + from openfeature.api import get_client + + self._run_immediate_handler(get_client(), event, handler) + + def remove_global_handler( + self, event: ProviderEvent, handler: EventHandler + ) -> None: + self._global_handlers[event].remove(handler) + + def run_handlers_for_provider( + self, + provider: FeatureProvider, + event: ProviderEvent, + provider_details: ProviderEventDetails, + ) -> None: + details = EventDetails.from_provider_event_details( + provider.get_metadata().name, provider_details + ) + # run the global handlers + self.run_global_handlers(event, details) + # run the handlers for clients associated to this provider + for client in self._client_handlers: + if client.provider == provider: + self.run_client_handlers(client, event, details) + + def _run_immediate_handler( + self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler + ) -> None: + if event == ProviderEvent.from_provider_status(client.get_provider_status()): + handler(EventDetails(provider_name=client.provider.get_metadata().name)) + + def clear(self) -> None: + self._global_handlers.clear() + self._client_handlers.clear() diff --git a/openfeature/api.py b/openfeature/api.py index 7704d382..bbbda081 100644 --- a/openfeature/api.py +++ b/openfeature/api.py @@ -1,10 +1,10 @@ import typing +from openfeature._event_support import EventSupport from openfeature.client import OpenFeatureClient from openfeature.evaluation_context import EvaluationContext from openfeature.event import ( EventHandler, - EventSupport, ProviderEvent, ProviderEventDetails, ) diff --git a/openfeature/event.py b/openfeature/event.py index f6d0b3a5..18ef4e1b 100644 --- a/openfeature/event.py +++ b/openfeature/event.py @@ -1,15 +1,11 @@ from __future__ import annotations -from collections import defaultdict from dataclasses import dataclass, field from enum import Enum -from typing import TYPE_CHECKING, Callable, ClassVar, Dict, List, Optional, Union +from typing import Callable, ClassVar, Dict, List, Optional, Union from openfeature.exception import ErrorCode -from openfeature.provider import FeatureProvider, ProviderStatus - -if TYPE_CHECKING: - from openfeature.client import OpenFeatureClient +from openfeature.provider import ProviderStatus class ProviderEvent(Enum): @@ -62,74 +58,3 @@ def from_provider_event_details( EventHandler = Callable[[EventDetails], None] - - -class EventSupport: - _global_handlers: Dict[ProviderEvent, List[EventHandler]] - _client_handlers: Dict[OpenFeatureClient, Dict[ProviderEvent, List[EventHandler]]] - - def __init__(self) -> None: - self._global_handlers = defaultdict(list) - self._client_handlers = defaultdict(lambda: defaultdict(list)) - - def run_client_handlers( - self, client: OpenFeatureClient, event: ProviderEvent, details: EventDetails - ) -> None: - for handler in self._client_handlers[client][event]: - handler(details) - - def run_global_handlers(self, event: ProviderEvent, details: EventDetails) -> None: - for handler in self._global_handlers[event]: - handler(details) - - def add_client_handler( - self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler - ) -> None: - handlers = self._client_handlers[client][event] - handlers.append(handler) - - self._run_immediate_handler(client, event, handler) - - def remove_client_handler( - self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler - ) -> None: - handlers = self._client_handlers[client][event] - handlers.remove(handler) - - def add_global_handler(self, event: ProviderEvent, handler: EventHandler) -> None: - self._global_handlers[event].append(handler) - - from openfeature.api import get_client - - self._run_immediate_handler(get_client(), event, handler) - - def remove_global_handler( - self, event: ProviderEvent, handler: EventHandler - ) -> None: - self._global_handlers[event].remove(handler) - - def run_handlers_for_provider( - self, - provider: FeatureProvider, - event: ProviderEvent, - provider_details: ProviderEventDetails, - ) -> None: - details = EventDetails.from_provider_event_details( - provider.get_metadata().name, provider_details - ) - # run the global handlers - self.run_global_handlers(event, details) - # run the handlers for clients associated to this provider - for client in self._client_handlers: - if client.provider == provider: - self.run_client_handlers(client, event, details) - - def _run_immediate_handler( - self, client: OpenFeatureClient, event: ProviderEvent, handler: EventHandler - ) -> None: - if event == ProviderEvent.from_provider_status(client.get_provider_status()): - handler(EventDetails(provider_name=client.provider.get_metadata().name)) - - def clear(self) -> None: - self._global_handlers.clear() - self._client_handlers.clear()