diff options
author | Yuqian Yang <crupest@crupest.life> | 2025-02-23 16:40:32 +0800 |
---|---|---|
committer | Yuqian Yang <crupest@crupest.life> | 2025-02-23 16:40:32 +0800 |
commit | 7e54b1ebe280d760347b1b640c9d4d0038ca6c58 (patch) | |
tree | 4d42c7a84066f0c2ae320e32dc7d86072fd0e4ae /services/manager/_event.py | |
parent | 34704b2090c48d9ab9e7458475fd701a38706ae7 (diff) | |
download | crupest-7e54b1ebe280d760347b1b640c9d4d0038ca6c58.tar.gz crupest-7e54b1ebe280d760347b1b640c9d4d0038ca6c58.tar.bz2 crupest-7e54b1ebe280d760347b1b640c9d4d0038ca6c58.zip |
feat(python): move python codes.
Diffstat (limited to 'services/manager/_event.py')
-rw-r--r-- | services/manager/_event.py | 61 |
1 files changed, 0 insertions, 61 deletions
diff --git a/services/manager/_event.py b/services/manager/_event.py deleted file mode 100644 index 51a794c..0000000 --- a/services/manager/_event.py +++ /dev/null @@ -1,61 +0,0 @@ -from __future__ import annotations - -from collections.abc import Callable -from typing import Generic, ParamSpec, TypeVar - -from .list import CruList - -_P = ParamSpec("_P") -_R = TypeVar("_R") - - -class CruEventHandlerToken(Generic[_P, _R]): - def __init__( - self, event: CruEvent, handler: Callable[_P, _R], once: bool = False - ) -> None: - self._event = event - self._handler = handler - self._once = once - - @property - def event(self) -> CruEvent: - return self._event - - @property - def handler(self) -> Callable[_P, _R]: - return self._handler - - @property - def once(self) -> bool: - return self._once - - -class CruEvent(Generic[_P, _R]): - def __init__(self, name: str) -> None: - self._name = name - self._tokens: CruList[CruEventHandlerToken] = CruList() - - def register( - self, handler: Callable[_P, _R], once: bool = False - ) -> CruEventHandlerToken: - token = CruEventHandlerToken(self, handler, once) - self._tokens.append(token) - return token - - def unregister(self, *handlers: CruEventHandlerToken | Callable[_P, _R]) -> int: - old_length = len(self._tokens) - self._tokens.reset( - self._tokens.as_cru_iterator().filter( - (lambda t: t in handlers or t.handler in handlers) - ) - ) - return old_length - len(self._tokens) - - def trigger(self, *args: _P.args, **kwargs: _P.kwargs) -> CruList[_R]: - results = CruList( - self._tokens.as_cru_iterator() - .transform(lambda t: t.handler(*args, **kwargs)) - .to_list() - ) - self._tokens.reset(self._tokens.as_cru_iterator().filter(lambda t: not t.once)) - return results |