Observer y Pub/Sub

Observer notifica a las partes interesadas cuando cambia el estado. Pub/Sub enruta mensajes a través de temas para que los publicadores no conozcan a los suscriptores. Python implementa ambos con llamables, list/set de manejadores, o librerías como blinker - no se requiere un framework pesado.

Receta

Tarjeta de receta de referencia rápida - lista para copiar y pegar.

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)

Cuándo usar esto:

Eventos de dominio - pedido realizado, usuario registrado, caché invalidada
Desacoplamiento de UI - la vista escucha, el modelo publica (escritorio, bucles de juego)
Hooks entre módulos - los plugins se suscriben sin importar los publicadores
Fan-out asíncrono - múltiples manejadores reaccionan a una señal
Pruebas - afirmar que los manejadores recibieron los eventos esperados

Ejemplo de Trabajo

Un servicio de pedidos publica eventos; los manejadores de auditoría y correo electrónico se suscriben de forma independiente.

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)

Lo que esto demuestra:

Los publicadores emiten eventos con nombre y carga útil de palabras clave.
Múltiples suscriptores por tema sin acoplamiento estrecho.
Copia de list() durante la emisión evita errores de mutación durante la iteración.
Los manejadores son funciones simples, fáciles de probar de forma aislada.

Inmersión Profunda

Cómo Funciona

Observer: el sujeto mantiene una lista de observadores; notify() llama a cada update()
Pub/Sub: el bus mapea cadenas de temas a listas de manejadores; los publicadores no tienen referencias a los suscriptores.
Los manejadores deben ser rápidos; el trabajo pesado va a colas o tareas en segundo plano.
unsubscribe elimina los manejadores para evitar fugas en buses de larga duración.
Los eventos tipados pueden usar dataclasses en lugar de diccionarios sueltos.

Observer vs Pub/Sub

Estilo	Acoplamiento	API Típica
Observer	El sujeto conoce la interfaz del observador	`.attach()` / `.notify()`
Pub/Sub	La cadena de tema media	`.subscribe(topic)` / `.publish()`

Notas de Python

# Referencias débiles para limpieza automática cuando el observador es recolectado por el GC
import weakref
 
class Subject:
    def __init__(self) -> None:
        self._observers: list = []
 
    def attach(self, obs):
        self._observers.append(weakref.ref(obs))

Trampas

Las excepciones del manejador abortan el fan-out - un suscriptor que falla bloquea a otros. Solución: envuelve cada llamada en try/except y registra.
Fugas de memoria - suscripciones olvidadas mantienen vivas las clausuras. Solución: anula la suscripción al desmontar o usa weakrefs.
Emisión reentrante - el manejador publica el mismo tema mientras maneja. Solución: documenta la reentrada o encola eventos.
Singleton de bus global - dependencias ocultas en todas partes. Solución: inyecta EventBus en los servicios.
Diccionarios de carga útil sin tipar - los errores tipográficos en las claves fallan silenciosamente. Solución: usa eventos dataclass o TypedDict.

Alternativas

Alternativa	Usar Cuando	No Usar Cuando
Llamadas directas a métodos	Dos colaboradores conocidos	Muchos suscriptores desconocidos
Colas `asyncio`	Trabajadores asíncronos consumen eventos	Suficientes callbacks síncronos simples
Broker de mensajes (Redis, Kafka)	Distribución entre procesos	Solo dentro del proceso
Señales (Django, blinker)	Integración con el framework	Se prefieren dependencias mínimas

Preguntas Frecuentes

¿Cuál es la diferencia entre Observer y Pub/Sub?

Observer se acopla a un objeto sujeto. Pub/Sub enruta por tema a través de un bus - los publicadores no referencian a los suscriptores directamente.

¿Deberían los manejadores ejecutarse de forma síncrona?

Para eventos de dominio dentro del proceso, generalmente sí - mantenlos rápidos. Descarga I/O a hilos, tareas o colas.

¿Cómo me desuscribo?

Devuelve un token desechable de subscribe, o pasa la misma referencia de función a unsubscribe.

¿Se admiten manejadores asíncronos?

Usa un bus asíncrono que await los manejadores de corrutinas, o programa tareas con asyncio.create_task.

¿Cómo ordeno la ejecución de los manejadores?

Documenta el orden FIFO de la lista de suscripción. Colas de prioridad si el orden importa para las reglas de negocio.

¿Qué pasa con los manejadores de una sola vez?

Envuelve con una bandera o usa el patrón functools.lru_cache - elimina el manejador después de la primera llamada.

¿Puedo usar tipado para eventos?

Sí. @dataclass(frozen=True) class OrderPlaced: order_id: str más emit(OrderPlaced(...)) mejora la seguridad.

¿Cómo pruebo los flujos de eventos?

Recopila eventos en una lista de manejadores de prueba. Afirma los nombres de los temas y las cargas útiles sin simular todo el bus.

¿Es esto lo mismo que las señales de Django?

La misma idea. Las señales de Django son pub/sub integradas en el framework; este patrón solo usa la biblioteca estándar.

¿Cuándo se convierte pub/sub en excesivo?

Dos clases con un grafo de llamadas claro - un método directo es más simple y fácil de rastrear.

Relacionado

Patrones Strategy y Dispatch - enruta por tipo de evento
Inyección de Dependencias en Python - inyecta el bus
Patrones Factory y Builder - construye servicios con buses
Conceptos Básicos de Patrones Pythonicos - resumen de la sección

Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+, y uv 0.6+.

Observer y Pub/Sub

Receta

Tarjeta de receta de referencia rápida - lista para copiar y pegar.

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)

Cuándo usar esto:

Eventos de dominio - pedido realizado, usuario registrado, caché invalidada
Desacoplamiento de UI - la vista escucha, el modelo publica (escritorio, bucles de juego)
Hooks entre módulos - los plugins se suscriben sin importar los publicadores
Fan-out asíncrono - múltiples manejadores reaccionan a una señal
Pruebas - afirmar que los manejadores recibieron los eventos esperados

Ejemplo de Trabajo

Un servicio de pedidos publica eventos; los manejadores de auditoría y correo electrónico se suscriben de forma independiente.

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)

Lo que esto demuestra:

Los publicadores emiten eventos con nombre y carga útil de palabras clave.
Múltiples suscriptores por tema sin acoplamiento estrecho.
Copia de list() durante la emisión evita errores de mutación durante la iteración.
Los manejadores son funciones simples, fáciles de probar de forma aislada.

Inmersión Profunda

Cómo Funciona

Observer: el sujeto mantiene una lista de observadores; notify() llama a cada update()
Pub/Sub: el bus mapea cadenas de temas a listas de manejadores; los publicadores no tienen referencias a los suscriptores.
Los manejadores deben ser rápidos; el trabajo pesado va a colas o tareas en segundo plano.
unsubscribe elimina los manejadores para evitar fugas en buses de larga duración.
Los eventos tipados pueden usar dataclasses en lugar de diccionarios sueltos.

Observer vs Pub/Sub

Estilo	Acoplamiento	API Típica
Observer	El sujeto conoce la interfaz del observador	`.attach()` / `.notify()`
Pub/Sub	La cadena de tema media	`.subscribe(topic)` / `.publish()`

Notas de Python

# Referencias débiles para limpieza automática cuando el observador es recolectado por el GC
import weakref
 
class Subject:
    def __init__(self) -> None:
        self._observers: list = []
 
    def attach(self, obs):
        self._observers.append(weakref.ref(obs))

Trampas

Las excepciones del manejador abortan el fan-out - un suscriptor que falla bloquea a otros. Solución: envuelve cada llamada en try/except y registra.
Fugas de memoria - suscripciones olvidadas mantienen vivas las clausuras. Solución: anula la suscripción al desmontar o usa weakrefs.
Emisión reentrante - el manejador publica el mismo tema mientras maneja. Solución: documenta la reentrada o encola eventos.
Singleton de bus global - dependencias ocultas en todas partes. Solución: inyecta EventBus en los servicios.
Diccionarios de carga útil sin tipar - los errores tipográficos en las claves fallan silenciosamente. Solución: usa eventos dataclass o TypedDict.

Alternativas

Alternativa	Usar Cuando	No Usar Cuando
Llamadas directas a métodos	Dos colaboradores conocidos	Muchos suscriptores desconocidos
Colas `asyncio`	Trabajadores asíncronos consumen eventos	Suficientes callbacks síncronos simples
Broker de mensajes (Redis, Kafka)	Distribución entre procesos	Solo dentro del proceso
Señales (Django, blinker)	Integración con el framework	Se prefieren dependencias mínimas

Preguntas Frecuentes

¿Cuál es la diferencia entre Observer y Pub/Sub?

Observer se acopla a un objeto sujeto. Pub/Sub enruta por tema a través de un bus - los publicadores no referencian a los suscriptores directamente.

¿Deberían los manejadores ejecutarse de forma síncrona?

Para eventos de dominio dentro del proceso, generalmente sí - mantenlos rápidos. Descarga I/O a hilos, tareas o colas.

¿Cómo me desuscribo?

Devuelve un token desechable de subscribe, o pasa la misma referencia de función a unsubscribe.

¿Se admiten manejadores asíncronos?

Usa un bus asíncrono que await los manejadores de corrutinas, o programa tareas con asyncio.create_task.

¿Cómo ordeno la ejecución de los manejadores?

Documenta el orden FIFO de la lista de suscripción. Colas de prioridad si el orden importa para las reglas de negocio.

¿Qué pasa con los manejadores de una sola vez?

Envuelve con una bandera o usa el patrón functools.lru_cache - elimina el manejador después de la primera llamada.

¿Puedo usar tipado para eventos?

Sí. @dataclass(frozen=True) class OrderPlaced: order_id: str más emit(OrderPlaced(...)) mejora la seguridad.

¿Cómo pruebo los flujos de eventos?

Recopila eventos en una lista de manejadores de prueba. Afirma los nombres de los temas y las cargas útiles sin simular todo el bus.

¿Es esto lo mismo que las señales de Django?

La misma idea. Las señales de Django son pub/sub integradas en el framework; este patrón solo usa la biblioteca estándar.

¿Cuándo se convierte pub/sub en excesivo?

Dos clases con un grafo de llamadas claro - un método directo es más simple y fácil de rastrear.

Relacionado

Patrones Strategy y Dispatch - enruta por tipo de evento
Inyección de Dependencias en Python - inyecta el bus
Patrones Factory y Builder - construye servicios con buses
Conceptos Básicos de Patrones Pythonicos - resumen de la sección

Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+, y uv 0.6+.

Observer y Pub/Sub

Buscar en la documentación

Observer y Pub/Sub