observer.py

from __future__ import annotations

from enum import IntEnum
from typing import Callable, Generic, List, Optional, Tuple, TypeVar

__all__ = ['Event', 'handler', 'EventMetaAction']


class EventMetaAction(IntEnum):
    handler_added = 1
    handler_removed = 2
    all_handlers_removed = 3
    deleted = 4
    announced = 5


T = TypeVar('T')
Handler = Callable[[T], None]
MetaT = Tuple[Optional[Handler[T]], Optional[T], EventMetaAction]

F = TypeVar('F', bound=Callable)


def handler(func: F) -> F:
    return func


class Event(Generic[T]):
    _handlers: List[Handler[T]]
    _meta: Optional[Event[MetaT]]

    def __init__(self) -> None:
        self._handlers = []
        self._meta = None

    @property
    def meta(self) -> Event[MetaT]:
        if self._meta is None:
            self._meta = Event()
        return self._meta

    def subscribe(self, handler: Handler[T]) -> None:
        self._handlers.append(handler)
        if self._meta is not None:
            self._meta.announce((handler, None, EventMetaAction.handler_added))

    def unsubscribe(self, handler: Handler[T]) -> None:
        try:
            self._handlers.remove(handler)
            if self._meta is not None:
                self._meta.announce((handler, None, EventMetaAction.handler_removed))
        except ValueError:
            pass

    def unsubscribe_all(self) -> None:
        self._handlers = []
        if self._meta is not None:
            self._meta.announce((None, None, EventMetaAction.all_handlers_removed))

    def announce(self, message: T) -> None:
        for sub in self._handlers:
            sub(message)
        if self._meta is not None:
            self._meta.announce((None, message, EventMetaAction.announced))

    def delete(self) -> None:
        if self._meta is not None:
            self._meta.announce((None, None, EventMetaAction.deleted))
        self._handlers = []
        if self._meta is not None:
            self._meta.delete()