Skip to content

Commit

Permalink
refactor: move EventSupport class to an internal module
Browse files Browse the repository at this point in the history
Signed-off-by: Federico Bond <federicobond@gmail.com>
  • Loading branch information
federicobond committed Mar 13, 2024
1 parent 870acad commit 2aba10e
Show file tree
Hide file tree
Showing 3 changed files with 89 additions and 78 deletions.
86 changes: 86 additions & 0 deletions openfeature/_event_support.py
Original file line number Diff line number Diff line change
@@ -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()
2 changes: 1 addition & 1 deletion openfeature/api.py
Original file line number Diff line number Diff line change
@@ -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,
)
Expand Down
79 changes: 2 additions & 77 deletions openfeature/event.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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()

0 comments on commit 2aba10e

Please sign in to comment.