Module rabbit.observer
View Source
import logging
from contextlib import suppress
import attr
from rabbit import loop
logging.getLogger(__name__).addHandler(logging.NullHandler())
@attr.s(slots=True, frozen=True)
class Observer:
_observers = attr.ib(
type=list, factory=list, validator=attr.validators.instance_of(list)
)
def attach(self, observer) -> None:
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer) -> None:
with suppress(ValueError):
self._observers.remove(observer)
def notify(self, modifier=None) -> None:
for observer in self._observers:
if modifier != observer:
logging.debug(f"{observer.__class__} notified")
loop().create_task(observer.configure())
def __len__(self):
return len(self._observers)
def __contains__(self, value):
if value in (self._observers):
return True
return False
@property
def observers(self):
return self._observers
Classes
Observer
class Observer(
observers: list = NOTHING
)
View Source
class Observer:
_observers = attr.ib(
type=list, factory=list, validator=attr.validators.instance_of(list)
)
def attach(self, observer) -> None:
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer) -> None:
with suppress(ValueError):
self._observers.remove(observer)
def notify(self, modifier=None) -> None:
for observer in self._observers:
if modifier != observer:
logging.debug(f"{observer.__class__} notified")
loop().create_task(observer.configure())
def __len__(self):
return len(self._observers)
def __contains__(self, value):
if value in (self._observers):
return True
return False
@property
def observers(self):
return self._observers
Instance variables
Methods
attach
def attach(
self,
observer
) -> None
View Source
def attach(self, observer) -> None:
if observer not in self._observers:
self._observers.append(observer)
detach
def detach(
self,
observer
) -> None
View Source
def detach(self, observer) -> None:
with suppress(ValueError):
self._observers.remove(observer)
notify
def notify(
self,
modifier=None
) -> None
View Source
def notify(self, modifier=None) -> None:
for observer in self._observers:
if modifier != observer:
logging.debug(f"{observer.__class__} notified")
loop().create_task(observer.configure())