Что такое паттерн наблюдатель?

Паттерн Наблюдатель (Observer Pattern)

Observer Pattern (Наблюдатель) — это поведенческий паттерн проектирования, который создаёт механизм подписки на события. Когда состояние объекта изменяется, все заинтересованные объекты (наблюдатели) автоматически уведомляются об этом. Паттерн реализует loose coupling — объекты слабо связаны между собой.

Основная архитектура

Subject (издатель): объект, который отслеживает и публикует события Observer (наблюдатель): объект, который слушает события и реагирует на них

Простой пример на чистом Python

from abc import ABC, abstractmethod
from typing import List

# Observer Interface
class Observer(ABC):
    """Интерфейс для наблюдателей"""
    
    @abstractmethod
    def update(self, subject):
        """Метод, вызываемый при изменении subject"""
        pass

# Subject
class User:
    """Издатель событий — пользователь"""
    
    def __init__(self, name: str, email: str):
        self._name = name
        self._email = email
        self._observers: List[Observer] = []
    
    @property
    def email(self):
        return self._email
    
    @email.setter
    def email(self, value: str):
        """При изменении email все наблюдатели уведомляются"""
        if self._email != value:
            self._email = value
            self.notify_observers()
    
    def subscribe(self, observer: Observer):
        """Добавить наблюдателя"""
        if observer not in self._observers:
            self._observers.append(observer)
    
    def unsubscribe(self, observer: Observer):
        """Удалить наблюдателя"""
        if observer in self._observers:
            self._observers.remove(observer)
    
    def notify_observers(self):
        """Оповестить всех наблюдателей об изменении"""
        for observer in self._observers:
            observer.update(self)

# Конкретные Observer'ы
class EmailService(Observer):
    """Сервис отправки писем"""
    
    def update(self, subject: User):
        print(f"[Email Service] Отправляем письмо на {subject.email}")

class LogService(Observer):
    """Сервис логирования"""
    
    def update(self, subject: User):
        print(f"[Log Service] Пользователь изменил email на {subject.email}")

class NotificationService(Observer):
    """Сервис уведомлений"""
    
    def update(self, subject: User):
        print(f"[Notification Service] Уведомление: email изменён")

# Использование
user = User("Иван", "ivan@example.com")
email_service = EmailService()
log_service = LogService()
notif_service = NotificationService()

user.subscribe(email_service)
user.subscribe(log_service)
user.subscribe(notif_service)

user.email = "newemail@example.com"
user.unsubscribe(email_service)
user.email = "another@example.com"

Пример с типизацией (Generic Observer)

from typing import TypeVar, Generic, Callable

T = TypeVar('T')

class Event(Generic[T]):
    """Событие с данными"""
    
    def __init__(self, data: T):
        self.data = data

class EventEmitter(Generic[T]):
    """Издатель событий"""
    
    def __init__(self):
        self._listeners: List[Callable[[Event[T]], None]] = []
    
    def on(self, listener: Callable[[Event[T]], None]):
        """Подписать слушателя на событие"""
        self._listeners.append(listener)
    
    def off(self, listener: Callable[[Event[T]], None]):
        """Отписать слушателя"""
        if listener in self._listeners:
            self._listeners.remove(listener)
    
    def emit(self, event: Event[T]):
        """Отправить событие всем слушателям"""
        for listener in self._listeners:
            listener(event)

# Использование
class Order:
    status_changed = EventEmitter[str]()
    
    def __init__(self, id: int):
        self.id = id
        self.status = "created"
    
    def update_status(self, new_status: str):
        self.status = new_status
        event = Event(new_status)
        self.status_changed.emit(event)

def on_order_status_changed(event: Event[str]):
    print(f"Статус заказа изменился на: {event.data}")

order = Order(1)
order.status_changed.on(on_order_status_changed)
order.update_status("processing")

Практический пример: бизнес-логика

class Product:
    """Издатель событий товара"""
    
    def __init__(self, name: str, price: float, stock: int):
        self.name = name
        self.price = price
        self.stock = stock
        self._observers: List[Observer] = []
    
    def subscribe(self, observer: Observer):
        self._observers.append(observer)
    
    def notify(self, event_type: str, data=None):
        for observer in self._observers:
            observer.handle_event(event_type, self, data)
    
    def set_price(self, new_price: float):
        if self.price != new_price:
            self.price = new_price
            self.notify("price_changed", new_price)
    
    def buy(self, quantity: int):
        if self.stock >= quantity:
            self.stock -= quantity
            self.notify("stock_decreased", quantity)
            if self.stock == 0:
                self.notify("out_of_stock")

class Observer(ABC):
    @abstractmethod
    def handle_event(self, event_type: str, subject: Product, data=None):
        pass

class PriceNotifier(Observer):
    def handle_event(self, event_type: str, subject: Product, data=None):
        if event_type == "price_changed":
            print(f"Цена изменилась на {data}")

class StockManager(Observer):
    def handle_event(self, event_type: str, subject: Product, data=None):
        if event_type == "stock_decreased":
            print(f"Осталось {subject.stock} шт.")
        elif event_type == "out_of_stock":
            print(f"Заказать новую партию!")

product = Product("Laptop", 1000, 5)
product.subscribe(PriceNotifier())
product.subscribe(StockManager())

product.set_price(900)
product.buy(5)

Observer Pattern в Django и Flask

# Django Signals
from django.db.models.signals import post_save
from django.dispatch import receiver
from myapp.models import User

@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
    if created:
        print(f"Новый пользователь: {instance.email}")

# Flask Events
from flask import Flask

app = Flask(__name__)

@app.before_request
def on_before_request():
    print("Перед обработкой запроса")

@app.after_request
def on_after_request(response):
    print("После обработки запроса")
    return response

Преимущества и недостатки

Преимущества:

• Слабая связанность между издателем и наблюдателями
• Динамическое добавление/удаление подписчиков во время выполнения
• Разделение ответственности между компонентами
• Поддержка broadcast-коммуникации

Недостатки:

• Сложность отладки — порядок уведомления непредсказуем
• Риск утечек памяти, если забыть отписать наблюдателя
• Производительность при большом количестве наблюдателей
• Может усложнить архитектуру при большом количестве событий

Observer Pattern — это идеальный выбор для систем событий, пользовательских интерфейсов и асинхронных операций в Python-приложениях.