6.7. Observer

• EN: Observer

• PL: Obserwator

• Type: object

6.7.1. Pattern

• When the state of the object changes and you need to notify other objects about this change

• Notify chart about changes in data to refresh

• Spreadsheet formulas

• Push or pull style of communication

6.7.2. Problem

6.7.3. Solution

from abc import ABC, abstractmethod
from dataclasses import dataclass, field


class Observer(ABC):
    @abstractmethod
    def update(self) -> None:
        pass

class Spreadsheet(Observer):
    def update(self) -> None:
        print('Spreadsheet got updated')

class Chart(Observer):
    def update(self) -> None:
        print('Chart got updated')


@dataclass
class Subject:
    """
    Observable - class which is observed
    """
    observers: list[Observer] = field(default_factory=list)

    def add_observer(self, observer: Observer) -> None:
        self.observers.append(observer)

    def remove_observer(self, observer: Observer) -> None:
        self.observers.remove(observer)

    def notify_observers(self):
        for observer in self.observers:
            observer.update()


class DataSource(Subject):
    value: int

    def get_value(self) -> int:
        return self.value

    def set_value(self, value) -> None:
        self.value = value
        self.notify_observers()


if __name__ == '__main__':
    datasource = DataSource()
    sheet1 = Spreadsheet()
    sheet2 = Spreadsheet()
    chart = Chart()

    datasource.add_observer(sheet1)
    datasource.add_observer(sheet2)
    datasource.add_observer(chart)

    datasource.set_value(1)

Code 6.39. push

from abc import ABC, abstractmethod
from dataclasses import dataclass, field


class Observer(ABC):
    @abstractmethod
    def update(self) -> None:
        pass


@dataclass
class Subject:
    """
    Observable - class which is observed
    """
    observers: list[Observer] = field(default_factory=list)

    def add_observer(self, observer: Observer) -> None:
        self.observers.append(observer)

    def remove_observer(self, observer: Observer) -> None:
        self.observers.remove(observer)

    def notify_observers(self):
        for observer in self.observers:
            observer.update()


class DataSource(Subject):
    value: int

    def get_value(self) -> int:
        return self.value

    def set_value(self, value) -> None:
        self.value = value
        self.notify_observers()


@dataclass
class Spreadsheet(Observer):
    datasource: DataSource

    def update(self) -> None:
        value = self.datasource.get_value()
        print(f'Spreadsheet got updated: {value}')


@dataclass
class Chart(Observer):
    datasource: DataSource

    def update(self) -> None:
        value = self.datasource.get_value()
        print(f'Chart got updated: {value}')


if __name__ == '__main__':
    datasource = DataSource()
    sheet1 = Spreadsheet(datasource)
    sheet2 = Spreadsheet(datasource)
    chart = Chart(datasource)

    datasource.add_observer(sheet1)
    datasource.add_observer(sheet2)
    datasource.add_observer(chart)

    datasource.set_value(1)

Code 6.40. pull

from abc import ABC, abstractmethod
from dataclasses import dataclass, field


class Observer(ABC):
    @abstractmethod
    def update(self, value: int) -> None:
        pass

class Spreadsheet(Observer):
    def update(self, value: int) -> None:
        print(f'Spreadsheet got updated: {value}')

class Chart(Observer):
    def update(self, value: int) -> None:
        print(f'Chart got updated: {value}')


@dataclass
class Subject:
    """
    Observable - class which is observed
    """
    observers: list[Observer] = field(default_factory=list)

    def add_observer(self, observer: Observer) -> None:
        self.observers.append(observer)

    def remove_observer(self, observer: Observer) -> None:
        self.observers.remove(observer)

    def notify_observers(self, value: int):
        for observer in self.observers:
            observer.update(value)


class DataSource(Subject):
    value: int

    def get_value(self) -> int:
        return self.value

    def set_value(self, value) -> None:
        self.value = value
        self.notify_observers(value)


if __name__ == '__main__':
    datasource = DataSource()
    sheet1 = Spreadsheet()
    sheet2 = Spreadsheet()
    chart = Chart()

    datasource.add_observer(sheet1)
    datasource.add_observer(sheet2)
    datasource.add_observer(chart)

    datasource.set_value(1)

6.7.4. Assignments