Шаблон Observer в Python

Что это такое

Паттерн Наблюдатель (Observer) позволяет одному объекту сообщать о событиях многим подписчикам, не зная, кто они. Подписчики регистрируются один раз; с этого момента каждое событие автоматически доходит до всех них.

Издатель зависит только от небольшого протокола подписчика — новые виды слушателей появляются без изменения хотя бы одной строки в издателе.

Пример: одно оповещение, каждому подписчику

Монитор хранит список подписчиков. Когда что-то происходит, он уведомляет весь список — сегодня это оповещение по электронной почте и панель мониторинга; завтра — сообщение в чат, достаточно просто подписать его.

Опубликовать один раз, доставить всем

Один вызов alert(), две доставки. Монитор понятия не имеет, что такое электронная почта или панель мониторинга, — он знает только о том, что можно уведомить.

Напишите программу на Python с паттерном Наблюдатель: протокол Subscriber с методом notify, подписчики EmailAlert и Dashboard и издатель Monitor с методами subscribe и alert, который уведомляет каждого подписчика о событии. Подпишите обоих и создайте одно оповещение.

> Вывод: полная программа на Python

Prompt

Напишите программу на Python с паттерном Наблюдатель: протокол Subscriber с методом notify, подписчики EmailAlert и Dashboard и издатель Monitor с методами subscribe и alert, который уведомляет каждого подписчика о событии. Подпишите обоих и создайте одно оповещение.

Вывод: полная программа на Python

from dataclasses import dataclass
from typing import Protocol

class Subscriber(Protocol):
  def notify(self, event: str) -> None: ...

@dataclass
class EmailAlert:
  addr: str

  def notify(self, event: str) -> None:
    print("письмо на", self.addr + ":", event)

class Dashboard:
  def notify(self, event: str) -> None:
    print("панель показывает:", event)

# Издатель хранит список и уведомляет всех, кто в нём есть.
class Monitor:
  def __init__(self) -> None:
    self.subscribers: list[Subscriber] = []

  def subscribe(self, subscriber: Subscriber) -> None:
    self.subscribers.append(subscriber)

  def alert(self, event: str) -> None:
    for subscriber in self.subscribers:
      subscriber.notify(event)

m = Monitor()
m.subscribe(EmailAlert(addr="[email protected]"))
m.subscribe(Dashboard())

m.alert("disk almost full")

Когда использовать

Используйте его, когда несколько частей программы должны реагировать на одни и те же события, а список реагирующих меняется, — обновления интерфейса, оповещения, инвалидация кэша, ленты активности.

Не используйте его, когда существует ровно один постоянный получатель, — вызывайте его напрямую. А в очень простых случаях обычный список функций обратного вызова даёт ту же форму паттерна вообще без механизма классов.

Плюсы и минусы

Плюсы: издатель и подписчики остаются слабо связанными; подписчики появляются и исчезают во время выполнения; новые реакции не требуют изменений в издателе.

Минусы: поток управления становится невидимым — чтобы ответить на вопрос «кто на это реагирует?», приходится искать; медленный или сбойный подписчик может застопорить цикл уведомлений; забытые отписки приводят к утечкам.