Simple demo of dynamically registering a provider to handle a specific event type

simple_provider_registration.py

import collections
import typing as t

class ServiceProvider:
    ...

class MessageProvider(t.Protocol):
    def write(self, message: str) -> None:
        ...

class PrintMessageProvider(MessageProvider):
    def write(self, message: str) -> None:
        print(message)

class FileMessageProvider(MessageProvider):
    def __init__(self, file_name: str):
        self._file_name = file_name

    def write(self, message: str) -> None:
        with open(self._file_name, "a+") as file:
            file.write(message)


_T = t.TypeVar("_T")


class ServiceManager:
    def __init__(self):
        self._providers: t.DefaultDict[t.Type[ServiceProvider], t.List[t.Any]] = collections.defaultdict(lambda: [])

    def register_service_provider(self, service_api: t.Type[ServiceProvider], provider: t.Type[t.Any]) -> None:
        # if doing multiple provider support
        # self._providers[service_api].append(provider)
        self._providers[service_api] = [provider]

    def get_service_provider(self, service_api: t.Type[_T]) -> _T:
        registered = self._providers.get(service_api)
        if not registered:
            raise Exception(f"No registered `{service_api.__name__}` service providers")

        return registered[0]


def task(service_manager: ServiceManager):
    try:
        msg_svc = service_manager.get_service_provider(MessageProvider)
        msg_svc.write("Hello from my provider!\n")
    except Exception as ex:
        print(ex)


manager = ServiceManager()


task(manager)


manager.register_service_provider(MessageProvider, PrintMessageProvider())
task(manager)


manager.register_service_provider(MessageProvider, FileMessageProvider("messages.txt"))
task(manager)