Observer & Pub/Sub

Observer notifica partes interessadas quando o estado muda. Pub/Sub roteia mensagens através de tópicos para que os publicadores não conheçam os assinantes. Python implementa ambos com chamáveis, list/set de manipuladores, ou bibliotecas como blinker - nenhum framework pesado é necessário.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

from collections import defaultdict
from typing import Callable
 
EventHandler = Callable[[str, dict], None]
 
class EventBus:
    def __init__(self) -> None:
        self._subs: dict[str, list[EventHandler]] = defaultdict(list)
 
    def subscribe(self, topic: str, handler: EventHandler) -> None:
        self._subs[topic].append(handler)
 
    def publish(self, topic: str, payload: dict) -> None:
        for handler in list(self._subs[topic]):
            handler(topic, payload)

Quando usar isso:

• Eventos de domínio - pedido realizado, usuário registrado, cache invalidado
• Desacoplamento de UI - a view escuta, o model publica (desktop, loops de jogo)
• Hooks entre módulos - plugins assinam sem importar publicadores
• Fan-out assíncrono - múltiplos manipuladores reagem a um único sinal
• Testes - afirmar que os manipuladores receberam os eventos esperados

Exemplo de Trabalho

Um serviço de pedidos publica eventos; manipuladores de auditoria e e-mail assinam independentemente.

from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable
 
Handler = Callable[[str, dict], None]
 
class EventBus:
    def __init__(self) -> None:
        self._handlers: dict[str, list[Handler]] = defaultdict(list)
 
    def on(self, event: str, handler: Handler) -> None:
        self._handlers[event].append(handler)
 
    def emit(self, event: str, **payload) -> None:
        for handler in list(self._handlers[event]):
            handler(event, payload)
 
@dataclass
class OrderService:
    bus: EventBus
    orders: list[dict] = field(default_factory=list)
 
    def place(self, sku: str, qty: int) -> dict:
        order = {"sku": sku, "qty": qty}
        self.orders.append(order)
        self.bus.emit("order.placed", order=order)
        return order
 
def audit_log(_event: str, payload: dict) -> None:
    print("AUDIT", payload)
 
def send_email(_event: str, payload: dict) -> None:
    print("EMAIL confirmation for", payload["order"]["sku"])
 
bus = EventBus()
bus.on("order.placed", audit_log)
bus.on("order.placed", send_email)
 
svc = OrderService(bus)
svc.place("BOOK-1", 2)

O que isso demonstra:

• Publicadores emitem eventos nomeados com carga útil de palavras-chave
• Múltiplos assinantes por tópico sem acoplamento rígido
• Cópia list() durante a emissão evita bugs de mutação durante iteração
• Manipuladores são funções simples - fáceis de testar isoladamente

Mergulho Profundo

Como Funciona

• Observer: o sujeito mantém uma lista de observadores; notify() chama cada update()
• Pub/Sub: o barramento mapeia strings de tópico para listas de manipuladores; publicadores não mantêm referências de assinantes
• Manipuladores devem ser rápidos; trabalho pesado vai para filas ou tarefas em segundo plano
• Desinscrever remove manipuladores para evitar vazamentos em barramentos de longa duração
• Eventos tipados podem usar dataclasses em vez de dicionários soltos

Observer vs Pub/Sub

Notas de Python

# Referências fracas para limpeza automática quando o observador é coletado pelo GC
import weakref
 
class Subject:
    def __init__(self) -> None:
        self._observers: list = []
 
    def attach(self, obs):
        self._observers.append(weakref.ref(obs))

Armadilhas

• Exceções de manipulador abortam o fan-out - um assinante falho bloqueia outros. Correção: envolva cada chamada em try/except e registre.
• Vazamentos de memória - assinaturas esquecidas mantêm closures vivas. Correção: desinscreva na desmontagem ou use weakrefs.
• Emissão reentrante - o manipulador publica o mesmo tópico durante o manuseio. Correção: documente a reentrância ou enfileire eventos.
• Singleton de barramento global - dependências ocultas em todos os lugares. Correção: injete EventBus em serviços.
• Dicionários de carga útil sem tipo - erros de digitação nas chaves falham silenciosamente. Correção: use eventos dataclass ou TypedDict.

Alternativas

FAQs

Relacionados

• Os Padrões Strategy & Dispatch - rotear por tipo de evento
• Injeção de Dependência em Python - injetar o barramento
• Padrões Factory & Builder - construir serviços com barramentos
• Noções Básicas de Padrões Pythonicos - visão geral da seção

Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.