Observer — Наблюдатель

Вы подписаны на YouTube-канал. Когда автор публикует новое видео — вы получаете уведомление. Вы не обязаны постоянно проверять канал вручную: платформа сама вас уведомит. Именно эту идею реализует паттерн Observer (Наблюдатель) — объект сообщает всем заинтересованным о своих изменениях автоматически.

📬 Аналогия: рассылка новостей

Прежде чем писать код, разберёмся с идеей на пальцах. Представьте, что есть интернет-магазин. У него есть блог с акциями. Люди могут подписаться на этот блог — и тогда при публикации новой акции они автоматически получат письмо, SMS или push-уведомление.

Ключевая особенность: издатель не знает, кто именно его слушает и что они с этим сделают. Кто-то отправит email, кто-то — SMS, кто-то — push. Издателю всё равно: он просто вызывает update() у каждого подписчика, а дальше — их дело.

😤 Проблема: жёсткая связь без Observer

Допустим, вы пишете класс Newsletter (новостная рассылка). Когда выходит новая статья, нужно уведомить подписчиков по email и SMS. Самое очевидное — вызвать нужные методы напрямую внутри Newsletter:

class EmailService:
    def send_email(self, message: str):
        print(f"[Email] Отправляем письмо: {message}")

class SMSService:
    def send_sms(self, message: str):
        print(f"[SMS] Отправляем SMS: {message}")


class Newsletter:
    def __init__(self):
        self.email_service = EmailService()
        self.sms_service = SMSService()

    def publish(self, article: str):
        print(f"Опубликована статья: {article}")
        # 🔴 ПЛОХО: Newsletter жёстко завязан на конкретные сервисы
        self.email_service.send_email(f"Новая статья: {article}")
        self.sms_service.send_sms(f"Новая статья: {article}")


newsletter = Newsletter()
newsletter.publish("10 советов по Python")

Паттерн Observer решает это радикально: издатель знает только об абстрактном интерфейсе Observer. Кто именно подписан — ему неважно.

🏗️ Структура паттерна Observer

Observer состоит из двух частей — Subject (издатель) и Observer (подписчик). Обе части описываются через абстрактные классы с помощью ABC.

Давайте запишем эту структуру в коде. Сначала — абстрактные классы:

from abc import ABC, abstractmethod
from typing import List


# ─────────────────────────────────────────────
# OBSERVER — абстрактный подписчик
# ─────────────────────────────────────────────
class Observer(ABC):
    """
    Интерфейс подписчика. Каждый, кто хочет
    получать уведомления, должен реализовать update().
    """

    @abstractmethod
    def update(self, subject: "Subject") -> None:
        """
        Вызывается издателем, когда происходит событие.
        subject — ссылка на издателя (можно взять его данные).
        """
        pass


# ─────────────────────────────────────────────
# SUBJECT — абстрактный издатель
# ─────────────────────────────────────────────
class Subject(ABC):
    """
    Интерфейс издателя. Умеет хранить и уведомлять подписчиков.
    """

    @abstractmethod
    def attach(self, observer: Observer) -> None:
        """Добавить подписчика в список."""
        pass

    @abstractmethod
    def detach(self, observer: Observer) -> None:
        """Удалить подписчика из списка."""
        pass

    @abstractmethod
    def notify(self) -> None:
        """Уведомить всех подписчиков — вызвать у каждого update()."""
        pass

📰 Реализуем издателя: класс Newsletter

Теперь напишем конкретный класс Newsletter — он наследуется от Subject и реализует все три метода. Плюс у него будет собственное состояние — текущая статья:

class Newsletter(Subject):
    """
    Конкретный издатель — новостная рассылка.
    Хранит список подписчиков и текущую статью.
    """

    def __init__(self) -> None:
        self._observers: List[Observer] = []  # список подписчиков
        self._latest_article: str = ""        # последняя опубликованная статья

    # ── Управление подписчиками ───────────────────

    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"  [+] Подписчик {observer.__class__.__name__} добавлен.")

    def detach(self, observer: Observer) -> None:
        try:
            self._observers.remove(observer)
            print(f"  [-] Подписчик {observer.__class__.__name__} удалён.")
        except ValueError:
            pass  # если подписчика не было — ничего не делаем

    def notify(self) -> None:
        """Обходим всех подписчиков и вызываем update() у каждого."""
        print(f"  [Newsletter] Уведомляем {len(self._observers)} подписчиков...")
        for observer in self._observers:
            observer.update(self)

    # ── Бизнес-логика ────────────────────────────

    @property
    def latest_article(self) -> str:
        """Подписчики смогут прочитать статью через это свойство."""
        return self._latest_article

    def publish(self, article: str) -> None:
        """Опубликовать новую статью и уведомить всех подписчиков."""
        print(f"\n📰 Опубликована статья: «{article}»")
        self._latest_article = article
        self.notify()  # автоматически уведомляем всех

👥 Конкретные подписчики: Email, SMS, Push

Теперь создадим несколько конкретных подписчиков. Каждый наследуется от Observer и реализует свой вариант update():

class EmailSubscriber(Observer):
    """Подписчик по email — получает статью на почту."""

    def __init__(self, email: str) -> None:
        self.email = email

    def update(self, subject: Subject) -> None:
        # Берём данные у издателя (pull model)
        article = subject.latest_article
        print(f"  📧 [Email → {self.email}] Новая статья: «{article}»")


class SMSSubscriber(Observer):
    """Подписчик по SMS — получает короткое уведомление."""

    def __init__(self, phone: str) -> None:
        self.phone = phone

    def update(self, subject: Subject) -> None:
        article = subject.latest_article
        # SMS обычно короткие — берём только первые 30 символов
        preview = article[:30] + "..." if len(article) > 30 else article
        print(f"  📱 [SMS → {self.phone}] Новость: {preview}")


class PushSubscriber(Observer):
    """Подписчик push-уведомлений — для мобильного приложения."""

    def __init__(self, device_id: str) -> None:
        self.device_id = device_id

    def update(self, subject: Subject) -> None:
        article = subject.latest_article
        print(f"  🔔 [Push → устройство {self.device_id}] 🔔 «{article}»")

🚀 Демонстрация: подписка, публикация, отписка

Соберём всё вместе и посмотрим на работу паттерна в действии:

# Создаём издателя
newsletter = Newsletter()

# Создаём подписчиков
email_user  = EmailSubscriber("alice@example.com")
sms_user    = SMSSubscriber("+7 900 123-45-67")
push_user   = PushSubscriber("device-abc-123")

# Подписываем
newsletter.attach(email_user)
newsletter.attach(sms_user)
newsletter.attach(push_user)

# Публикуем первую статью — все три получат уведомление
newsletter.publish("10 советов по Python для начинающих")

# Отписываем SMS-подписчика
newsletter.detach(sms_user)

# Публикуем вторую статью — SMS уже не получит уведомление
newsletter.publish("Как работают декораторы в Python")

Вывод программы:

  [+] Подписчик EmailSubscriber добавлен.
  [+] Подписчик SMSSubscriber добавлен.
  [+] Подписчик PushSubscriber добавлен.

📰 Опубликована статья: «10 советов по Python для начинающих»
  [Newsletter] Уведомляем 3 подписчиков...
  📧 [Email → alice@example.com] Новая статья: «10 советов по Python для начинающих»
  📱 [SMS → +7 900 123-45-67] Новость: 10 советов по Python для...
  🔔 [Push → устройство device-abc-123] 🔔 «10 советов по Python для начинающих»

  [-] Подписчик SMSSubscriber удалён.

📰 Опубликована статья: «Как работают декораторы в Python»
  [Newsletter] Уведомляем 2 подписчиков...
  📧 [Email → alice@example.com] Новая статья: «Как работают декораторы в Python»
  🔔 [Push → устройство device-abc-123] 🔔 «Как работают декораторы в Python»

💻 Полный код: всё вместе

Соберём финальную версию в одном месте — удобно скопировать и запустить:

from abc import ABC, abstractmethod
from typing import List


# ══════════════════════════════════════════════
# АБСТРАКТНЫЕ КЛАССЫ (интерфейсы)
# ══════════════════════════════════════════════

class Observer(ABC):
    @abstractmethod
    def update(self, subject: "Subject") -> None:
        pass


class Subject(ABC):
    @abstractmethod
    def attach(self, observer: Observer) -> None:
        pass

    @abstractmethod
    def detach(self, observer: Observer) -> None:
        pass

    @abstractmethod
    def notify(self) -> None:
        pass


# ══════════════════════════════════════════════
# КОНКРЕТНЫЙ ИЗДАТЕЛЬ
# ══════════════════════════════════════════════

class Newsletter(Subject):
    def __init__(self) -> None:
        self._observers: List[Observer] = []
        self._latest_article: str = ""

    def attach(self, observer: Observer) -> None:
        if observer not in self._observers:
            self._observers.append(observer)
            print(f"  [+] {observer.__class__.__name__} подписан.")

    def detach(self, observer: Observer) -> None:
        try:
            self._observers.remove(observer)
            print(f"  [-] {observer.__class__.__name__} отписан.")
        except ValueError:
            pass

    def notify(self) -> None:
        for observer in self._observers:
            observer.update(self)

    @property
    def latest_article(self) -> str:
        return self._latest_article

    def publish(self, article: str) -> None:
        print(f"\n📰 Публикуем: «{article}»")
        self._latest_article = article
        self.notify()


# ══════════════════════════════════════════════
# КОНКРЕТНЫЕ ПОДПИСЧИКИ
# ══════════════════════════════════════════════

class EmailSubscriber(Observer):
    def __init__(self, email: str) -> None:
        self.email = email

    def update(self, subject: Subject) -> None:
        print(f"  📧 Email → {self.email}: «{subject.latest_article}»")


class SMSSubscriber(Observer):
    def __init__(self, phone: str) -> None:
        self.phone = phone

    def update(self, subject: Subject) -> None:
        preview = subject.latest_article[:30]
        print(f"  📱 SMS → {self.phone}: {preview}...")


class PushSubscriber(Observer):
    def __init__(self, device_id: str) -> None:
        self.device_id = device_id

    def update(self, subject: Subject) -> None:
        print(f"  🔔 Push → {self.device_id}: «{subject.latest_article}»")


# ══════════════════════════════════════════════
# ИСПОЛЬЗОВАНИЕ
# ══════════════════════════════════════════════

if __name__ == "__main__":
    newsletter = Newsletter()

    alice = EmailSubscriber("alice@example.com")
    bob   = SMSSubscriber("+7 900 123-45-67")
    carol = PushSubscriber("device-abc-123")

    # Все подписываются
    newsletter.attach(alice)
    newsletter.attach(bob)
    newsletter.attach(carol)

    # Первая статья — получат все трое
    newsletter.publish("Введение в ООП на Python")

    # Bob отписывается
    newsletter.detach(bob)

    # Вторая статья — получат Alice и Carol
    newsletter.publish("Паттерн Observer на практике")

⚡ Push-модель vs Pull-модель

Существуют два способа передачи данных от издателя подписчику. Давайте разберём оба:

def notify(self):
    for obs in self._observers:
        obs.update(self._latest_article)

def update(self, article: str):
    print(f"Получили: {article}")

def notify(self):
    for obs in self._observers:
        obs.update(self)  # self = весь издатель

def update(self, subject):
    # Берём только что нужно
    data = subject.latest_article

В нашем примере мы использовали pull-модель — передаём subject, и каждый подписчик сам берёт нужные данные. Это более гибкий подход: один подписчик возьмёт latest_article, другой — author, третий — published_at, не меняя сигнатуру notify().

🌍 Observer в реальном мире

Паттерн Observer — один из самых распространённых. Вы уже используете его, возможно не зная об этом:

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        print(f"Новый пользователь: {instance.username}. Отправляем приветствие!")

# User (Subject) автоматически вызовет send_welcome_email (Observer)
# после каждого сохранения — без изменения кода модели User

Видите? Модель User не знает о функции send_welcome_email. Она просто сигналит «я сохранена», а зарегистрированные обработчики реагируют. Это Observer в действии в реальном Django-проекте.

🎯 Когда применять Observer?

🎯 Итоги урока