Module peprock.patterns
Reusable software design patterns.
Expand source code
"""Reusable software design patterns."""
from .observer import Observer, Subject
__all__ = [
"Subject",
"Observer",
]
Sub-modules
peprock.patterns.observer
-
Subject notifies observers of state changes …
Classes
class Observer
-
Receive notifications from subjects after registration.
Expand source code
class Observer(typing.Generic[_P], metaclass=abc.ABCMeta): """Receive notifications from subjects after registration.""" @abc.abstractmethod def notify( self: Observer[_P], __subject: Subject[_P], *args: _P.args, **kwargs: _P.kwargs, ) -> None: """Notify of state change of subject.""" ...
Ancestors
- typing.Generic
Methods
def notify(self: Observer[_P], _Observer__subject: Subject[_P], *args: _P.args, **kwargs: _P.kwargs) ‑> None
-
Notify of state change of subject.
Expand source code
@abc.abstractmethod def notify( self: Observer[_P], __subject: Subject[_P], *args: _P.args, **kwargs: _P.kwargs, ) -> None: """Notify of state change of subject.""" ...
class Subject
-
Notify subjects of state changes and manage observer registration.
Expand source code
class Subject(typing.Generic[_P]): """Notify subjects of state changes and manage observer registration.""" @functools.cached_property def _observers(self: Subject[_P]) -> weakref.WeakSet[Observer[_P]]: return weakref.WeakSet() def register_observer(self: Subject[_P], __observer: Observer[_P], /) -> None: """Register observer.""" self._observers.add(__observer) def unregister_observer(self: Subject[_P], __observer: Observer[_P], /) -> None: """Unregister observer.""" self._observers.discard(__observer) def is_registered_observer(self: Subject[_P], __observer: Observer[_P], /) -> bool: """Check if observer is registered and return as bool.""" return __observer in self._observers def notify_observers( self: Subject[_P], *args: _P.args, **kwargs: _P.kwargs, ) -> None: """Notify registered observers by calling their notify() method.""" for observer in self._observers: # noinspection PyTypeChecker observer.notify(self, *args, **kwargs)
Ancestors
- typing.Generic
Methods
def is_registered_observer(self: Subject[_P], _Subject__observer: Observer[_P], /) ‑> bool
-
Check if observer is registered and return as bool.
Expand source code
def is_registered_observer(self: Subject[_P], __observer: Observer[_P], /) -> bool: """Check if observer is registered and return as bool.""" return __observer in self._observers
def notify_observers(self: Subject[_P], *args: _P.args, **kwargs: _P.kwargs) ‑> None
-
Notify registered observers by calling their notify() method.
Expand source code
def notify_observers( self: Subject[_P], *args: _P.args, **kwargs: _P.kwargs, ) -> None: """Notify registered observers by calling their notify() method.""" for observer in self._observers: # noinspection PyTypeChecker observer.notify(self, *args, **kwargs)
def register_observer(self: Subject[_P], _Subject__observer: Observer[_P], /) ‑> None
-
Register observer.
Expand source code
def register_observer(self: Subject[_P], __observer: Observer[_P], /) -> None: """Register observer.""" self._observers.add(__observer)
def unregister_observer(self: Subject[_P], _Subject__observer: Observer[_P], /) ‑> None
-
Unregister observer.
Expand source code
def unregister_observer(self: Subject[_P], __observer: Observer[_P], /) -> None: """Unregister observer.""" self._observers.discard(__observer)