Design Patterns: Il Pattern Observer
Pubblicato da Michele Saba
Introduzione e il perché di questo argomento
Oggi tratteremo un argomento che va controcorrente: design pattern.
Introduciamo il pattern Observer, e soprattutto perché è utile nel mondo reale.
Oggi abbiamo strumenti velocissimi per produrre codice, ma la velocità sposta il problema: non è complicato iniziare a far funzionare qualcosa, è complicato farla crescere bene e farla diventare un sistema reale e affidabile.
E quando un progetto cresce, l’accoppiamento e la mancanza di struttura si trasformano in costi: modifiche che rompono altro, test che diventano impossibili, e feature “banali” che richiedono di toccare dieci punti diversi.
I design pattern non sono regole, non sono dogmi, non sono algoritmi: sono soluzioni ricorrenti a problemi comuni. Non sono qualcosa da copiare e incollare, ma concetti generali per risolvere problemi particolari.
E Observer è proprio questo: uno schema pulito per far succedere “cose” quando accade un evento… senza implementare tutta la logica nello stesso punto.
Classe YouTubeChannel: Implementazione ingenua
Partiamo con un esempio pratico: supponiamo di modellare la realtà di un canale YouTube. Questo pubblica un video, e quel punto succedono cose diverse: email, notifiche agli iscritti, log, analytics…
Un'implementazione ingenua può funzionare, ma si degrada velocemente.
Nel nostro esempio di codice abbiamo una classe YouTubeChannel che crea un evento VideoPublished quando un video è pubblicato.
Procede quindi a chiamare tutti i metodi a sua disposizione che sono in essa definiti per gestire le varie operazioni del caso.
Il tutto avviene dentro al suo metodo publish().
from dataclasses import dataclass
@dataclass(frozen=True)
class VideoPublished:
"""
Evento generato quando un video viene pubblicato.
"""
video_id: str
title: str
class YoutubeChannel:
"""
YoutubeChannel è il soggetto che pubblica l'evento.
"""
def publish(self, video_id: str, title: str) -> None:
event = VideoPublished(video_id, title)
self._send_email_notifications(event)
self._send_push_notifications(event)
self._update_search_index(event)
def _send_email_notifications(self, event: VideoPublished) -> None:
print(f"[email] New video: {event.title}")
def _send_push_notifications(self, event: VideoPublished) -> None:
print(f"[push] New video: {event.title}")
def _update_search_index(self, event: VideoPublished) -> None:
print(f"[index] Indexed: {event.video_id}")Da questo codice è evidente che la complessità del componente aumenterà da subito.
Per inviare notifiche dovremmo anzitutto conoscere chi sono gli iscritti, e ci servirà un sistema per inviare le mail, e magari vogliamo aggiungere un sistema di invio notifiche anche su Telegram.
La nostra classe è appena nata e sta già per esplodere, caricata di responsabilità che non le competono:
from dataclasses import dataclass
class EmailService:
"""
Il servizio che invia le email.
"""
def send_new_video_email(self, user_ids: list[str], title: str) -> None:
for user_id in user_ids:
print(f"[email → {user_id}] New video published: {title}")
@dataclass(frozen=True)
class VideoPublished:
"""
Evento generato quando un video viene pubblicato.
"""
video_id: str
title: str
class YoutubeChannel:
"""
YoutubeChannel è il soggetto che pubblica l'evento.
"""
def __init__(self) -> None:
# Hard-coded infrastructure dependencies
self._email_service = EmailService()
def publish(self, video_id: str, title: str) -> None:
event = VideoPublished(video_id, title)
# Business logic
self._send_email_notifications(event)
self._send_push_notifications(event)
self._update_search_index(event)
self._post_to_telegram(event)
def _send_email_notifications(self, event: VideoPublished) -> None:
subscriber_ids = self._load_subscriber_ids()
self._email_service.send_new_video_email(
user_ids=subscriber_ids,
title=event.title,
)
def _load_subscriber_ids(self) -> list[str]:
# query al DB
return ["u123", "u456", "u789"]
def _update_search_index(self, event: VideoPublished) -> None:
print(f"[index] Indexed video {event.video_id}")
def _send_push_notifications(self, event: VideoPublished) -> None:
print(f"[push] New video: {event.title}")
def _post_to_telegram(self, event: VideoPublished) -> None:
print(f"[telegram] New video: {event.title}")Cos'è il pattern Observer e a cosa serve?
Il pattern Observer descrive un meccanismo in cui un oggetto centrale segnala che qualcosa è successo a chi ha scelto di ricevere aggiornamenti, così questi possono reagire senza che l’oggetto centrale debba conoscere cosa faranno.
In altre parole: il soggetto emette un evento, gli observer reagiscono, e le due parti restano disaccoppiate.
Definiamo una classe Subscriber, che definisce un "modello" del nostro Osservatore ideale. Per noi è sufficiente che implementi un metodo notify(), così che sia possibile contattarlo. Deciderà poi il Subscriber come comportarsi e che azioni intraprendere.
Se non avete mai sentito parlare di Protocol, si tratta di un sistema che serve a descrivere un comportamento. Con un Protocol possiamo dire:
“Non ci interessa che classe sei, ci interessa che metodi hai.”
Nel nostro caso: qualsiasi oggetto che espone notify(event) è un Subscriber, anche senza ereditarlo.
Volendo avremmo infatti potuto usare anche le Abstract Base Classes di Python, ma un Subscriber così definito resta più vincolante.
Nel nostro caso gli Observer (osservatori) saranno quindi oggetti Subscriber, e l'oggetto osservato che emette gli eventi è la classe YouTubeChannel.
Come usare il pattern Observer in Python (esempio pratico)
from dataclasses import dataclass
from typing import Protocol
@dataclass(frozen=True)
class VideoPublished:
"""
Evento generato quando un video viene pubblicato.
"""
video_id: str
title: str
class Subscriber(Protocol):
"""
Gli observer devono implementare il metodo notify.
In questo modo, possiamo aggiungere observer di tipo diverso alla lista, senza che il soggetto ne sappia nulla.
"""
def notify(self, event: VideoPublished) -> None: ...
class YoutubeChannel:
"""
YoutubeChannel è il soggetto che pubblica l'evento.
Responsabilità chiara: il soggetto emette, gli observer reagiscono.
"""
def __init__(self) -> None:
"""
Inizializza la lista degli observer.
"""
self._subs: list[Subscriber] = []
def subscribe(self, sub: Subscriber) -> None:
"""
Aggiunge un observer alla lista.
Controllando che l'observer implementi il metodo notify, possiamo
facilmente impiegare observer esterni, come classi esterne alla nostra applicazione.
"""
if not callable(getattr(sub, "notify", None)):
raise TypeError("Subscriber must have a callable notify(event) method")
if sub in self._subs:
return
self._subs.append(sub)
def unsubscribe(self, sub: Subscriber) -> None:
"""
Rimuove un observer dalla lista.
"""
try:
self._subs.remove(sub)
except ValueError:
pass
def publish(self, video_id: str, title: str) -> None:
"""
Pubblica un evento e notifica gli observer tramite il loro metodo notify.
Facciamo uno "snapshot" della lista degli observer per evitare inconvenienti se un
observer si iscrive/disiscrive durante la notifica.
"""
event = VideoPublished(video_id, title)
for s in list(self._subs):
try:
s.notify(event)
except Exception as e:
print(f"Error notifying subscriber {s}: {e}")Abbiamo ora la possibilità di creare varie classi Observer per implementare correttamente la nostra logica di business, ottenendo come vantaggi:
• Responsabilità chiara: YoutubeChannel crea gli eventi e li emette, non reagisce.
• Disaccoppiamento: il soggetto non conosce i dettagli sul come l'evento verrà gestito
• Open/Closed rispettato: nuovi comportamenti = nuovi observer, senza modifiche al soggetto.
L'idea di un Observer dedicato alla gestione email ha ad esempio perfettamente senso.
Possiamo interfacciarci con diversi servizi di invio tramite composizione, caricare la lista degli iscritti da una sorgente esterna (nel nostro caso simuliamo una query al nostro DB), e far crescere la classe secondo le nostre esigenze, il tutto senza mai dover toccare la classe YouTubeChannel. Finché esponiamo un metodo notify() siamo apposto!
class EmailService:
"""
Il servizio che invia le email.
"""
def send_new_video_email(self, user_ids: list[str], title: str) -> None:
for user_id in user_ids:
print(f"[email → {user_id}] New video published: {title}")
class EmailNotifications:
"""
Observer che invia notifiche via email.
"""
def __init__(self, email_service: EmailService) -> None:
self._email_service = email_service
def notify(self, event: VideoPublished) -> None:
self._send_email_notifications(event)
def _send_email_notifications(self, event: VideoPublished) -> None:
subscriber_ids = self._load_subscriber_ids()
self._email_service.send_new_video_email(
user_ids=subscriber_ids,
title=event.title,
)
def _load_subscriber_ids(self) -> list[str]:
# simuliamo una query al DB
return ["u123", "u456", "u789"]Aggiungiamo ora qualche altro Observer
class PushNotifications:
def notify(self, event: VideoPublished) -> None:
print(f"[push] New video: {event.title}")
class SearchIndexUpdater:
def notify(self, event: VideoPublished) -> None:
print(f"[index] Indexed: {event.video_id}")
class TelegramPoster:
def notify(self, event: VideoPublished) -> None:
print(f"[telegram] New video: {event.title}")Non ci resta che provare ad eseguire il codice!
Inizializziamo il nostro canale in una variabile channel, e creiamo dei nuovi oggetti Subscriber di natura diversa.
Usiamo i metodi subscribe() e unsubscribe() per gestire il flusso di iscrizione.
Ci basta ora decidere di pubblicare un video dal canale per notificare tutti gli iscritti
if __name__ == "__main__":
channel = YoutubeChannel()
email_service = EmailService()
email_subscriber = EmailNotifications(email_service)
channel.subscribe(email_subscriber)
push_subscriber = PushNotifications()
channel.subscribe(push_subscriber)
search_index_subscriber = SearchIndexUpdater()
channel.subscribe(search_index_subscriber)
telegram_subscriber = TelegramPoster()
channel.subscribe(telegram_subscriber)
channel.publish(video_id="xyz123", title="Design Pattern: Observer (video anti slop)")
# output:
# [email → u123] New video published: Design Pattern: Observer (video anti slop)
# [email → u456] New video published: Design Pattern: Observer (video anti slop)
# [email → u789] New video published: Design Pattern: Observer (video anti slop)
# [push] New video: Design Pattern: Observer (video anti slop)
# [index] Indexed: xyz123
# [telegram] New video: Design Pattern: Observer (video anti slop)In chiusura, come avete notato il pattern Observer è potente, ma attenzione: se abusato, può rendere difficile seguire il flusso logico del programma perché le azioni avvengono sempre "dietro le quinte"!
Approfondimenti
Un iscritto al canale YouTube (quello vero, dove ho pubblicato questo video) fa una domanda interessante. La pubblico qui, assieme alla mia risposta:
Domanda: "Molto interessante! Forse mi avrebbe fatto piacere capire meglio i benefici di Protocol nell’implementare questa strategia: che differenza ci sarebbe stata non utilizzandolo?
Risposta: "Ciao, felice che il video ti sia piaciuto, domanda interessante.
Il vantaggio principale di Protocol non è a runtime, ma a livello di design e type checking.
Permette di definire un contratto strutturale, senza imporre ereditarietà (come nel caso delle Abstract Base Classes). In questo modo oggetti esterni possono diventare subscriber senza dover essere adattati, perché fintanto che espongono un metodo notify() a noi vanno bene.
Con Protocol veniamo avvisati dai tool di type checking di eventuali errori in fase di sviluppo (assenza di notify() nel nostro caso) e quindi teoricamente preveniamo errori a runtime. Non solo, siamo molto più espliciti a livello di intenzioni per chi legge il codice.
Senza Protocol avrei avuto come alternative:
1) nessun contratto: più flessibile, ma meno sicuro e meno esplicito
2) ABC / classe base: approccio più rigido e invasivo; da quanto è uscito Protocol si opta principalmente per questo
Un saluto!