domain_event.py 633 B

1234567891011121314151617181920212223242526
  1. # -*- coding: utf-8 -*-
  2. # (c) Nelen & Schuurmans
  3. from abc import ABC
  4. from typing import Awaitable, Callable, TypeVar
  5. import blinker
  6. __all__ = ["DomainEvent"]
  7. TDomainEvent = TypeVar("TDomainEvent", bound="DomainEvent")
  8. TEventHandler = Callable[[TDomainEvent], Awaitable[None]]
  9. class DomainEvent(ABC):
  10. @classmethod
  11. def _signal(cls) -> blinker.Signal:
  12. return blinker.signal(cls.__name__)
  13. @classmethod
  14. def register_handler(cls, receiver: TEventHandler) -> TEventHandler:
  15. return cls._signal().connect(receiver)
  16. async def send_async(self) -> None:
  17. await self._signal().send_async(self)