

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

observers

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())