Introduzione
Quando si parla di programmazione asincrona in Python, spesso l’attenzione si concentra su coroutine, task e concorrenza. Un aspetto molto pratico, ma spesso sottovalutato, è però la gestione efficiente degli stream di eventi: file che arrivano a pezzi, dati di rete, messaggi prodotti da sensori, log in tempo reale o flussi di testo da processare senza bloccare il programma.
In questi casi, async e await permettono di leggere e processare i dati man mano che arrivano, evitando di caricare tutto in memoria o di bloccare l’intero processo in attesa di un’operazione lenta. È un approccio ideale quando devi costruire applicazioni reattive, come server TCP, client di streaming, pipeline di elaborazione dati o tool che monitorano risorse in tempo reale.
In questo tutorial vedremo un sotto-argomento pratico e molto utile: come leggere e processare uno stream di dati in modo asincrono, usando un esempio originale basato su un flusso di righe testuali che rappresentano eventi di telemetria.
Codice completo
import asyncio
from dataclasses import dataclass
from typing import AsyncIterator
@dataclass
class TelemetryEvent:
source: str
value: float
async def fake_stream() -> AsyncIterator[str]:
"""
Simula uno stream di dati che arriva in modo irregolare.
Ogni riga rappresenta un evento nel formato:
source,value
"""
raw_lines = [
"sensor-a,12.5",
"sensor-b,18.2",
"sensor-a,13.1",
"sensor-c,9.8",
"sensor-b,19.0",
"sensor-a,14.4",
]
for line in raw_lines:
await asyncio.sleep(0.4) # Simula latenza di rete o I/O
yield line
async def parse_event(line: str) -> TelemetryEvent:
"""
Converte una riga CSV molto semplice in un evento tipizzato.
"""
source, value = line.split(",")
return TelemetryEvent(source=source, value=float(value))
async def process_event(event: TelemetryEvent) -> None:
"""
Elabora un singolo evento.
In un caso reale qui potresti:
- salvare su database
- inviare a un´API
- aggiornare una dashboard
- calcolare statistiche incrementali
"""
await asyncio.sleep(0.2) # Simula un´operazione I/O-bound
print(f"[OK] Evento ricevuto da {event.source}: {event.value}")
async def consume_stream() -> None:
"""
Legge lo stream evento per evento, senza bloccare il programma.
"""
async for line in fake_stream():
event = await parse_event(line)
await process_event(event)
async def main() -> None:
print("Avvio consumo stream asincrono...")
await consume_stream()
print("Stream terminato.")
if __name__ == "__main__":
asyncio.run(main())Spiegazione
Il cuore dell’esempio è la funzione fake_stream(), che simula una sorgente dati asincrona. Invece di restituire una lista completa, produce i dati uno alla volta con yield e introduce una pausa con await asyncio.sleep(). Questo comportamento è simile a ciò che accade quando leggi dati da una socket, da una coda o da un file remoto.
La funzione consume_stream() usa async for per iterare sullo stream. Questa è una differenza importante rispetto al classico for: qui l’iterazione non avviene su una sequenza già pronta, ma su un iteratore asincrono che può sospendere l’esecuzione in attesa del prossimo dato.
Vediamo i passaggi principali:
- fake_stream() genera righe testuali in modo graduale.
- parse_event() trasforma una stringa in un oggetto strutturato TelemetryEvent.
- process_event() simula un’operazione lenta ma asincrona, come una scrittura su database o una richiesta HTTP.
- consume_stream() coordina il flusso: legge, converte e processa ogni evento.
Un dettaglio importante è che tutto questo avviene senza bloccare l’event loop. Se il programma dovesse gestire anche altri task, questi potrebbero continuare a girare mentre lo stream attende nuovi dati o mentre un evento viene elaborato.
Questo approccio è utile soprattutto per operazioni I/O-bound, cioè legate all’attesa di input/output. Non è invece la scelta migliore per calcoli pesanti su CPU: in quel caso servono altre tecniche, come multiprocessing o delega a thread/processi separati.
Un altro punto interessante è l’uso di dataclass. Invece di lavorare con tuple o dizionari anonimi, definiamo un modello chiaro per i dati ricevuti. Questo migliora leggibilità, manutenzione e sicurezza del codice.
Best practice
- Usa async for quando lo stream è davvero asincrono: ad esempio file remoti, socket, generatori asincroni o code di messaggi.
- Separare parsing e processing: trasformare i dati in un formato strutturato prima di elaborarli rende il codice più testabile.
- Evita operazioni bloccanti dentro le coroutine: chiamate lente non asincrone possono fermare l’intero event loop.
- Tipizza i dati con dataclass o classi dedicate: aiuta a ridurre errori e migliora la comprensione del flusso.
- Gestisci gli errori vicino al punto in cui possono accadere: ad esempio durante il parsing, se i dati in ingresso non sono validi.
- Non accumulare tutto in memoria se puoi elaborare evento per evento: è uno dei principali vantaggi dell’asincronia sugli stream.
Se vuoi rendere l’elaborazione più efficiente, puoi anche introdurre una pipeline con più task: uno legge, uno valida, uno persiste. In questo modo separi le responsabilità e sfrutti meglio la concorrenza, purché i passaggi siano progettati con attenzione.
Riepilogo
La programmazione asincrona in Python non serve solo a “fare cose in parallelo”, ma soprattutto a gestire bene le attese. Lavorare con stream di eventi è un caso d’uso perfetto per async e await, perché permette di processare i dati appena arrivano, senza bloccare il resto dell’applicazione.
In questo tutorial hai visto come:
- creare uno stream asincrono con un generatore;
- iterare i dati con async for;
- separare parsing e processing;
- usare asyncio.run() per avviare il programma;
- strutturare il codice in modo leggibile e manutenibile.
Questo pattern è molto utile in applicazioni reali come monitoraggio eventi, ingestion di log, pipeline di telemetria e servizi di integrazione con API che producono dati progressivamente.
Approfondisci con risorse ufficiali
- Documentazione ufficiale di asyncio: panoramica completa su event loop, task, coroutine e stream.
- PEP 492: introduce async e await in Python.
- PEP 525: spiega gli async generator, utili per creare stream asincroni.
- Documentazione di dataclasses: per modellare dati in modo semplice e pulito.
Se vuoi fare un passo in più, il prossimo argomento naturale è costruire una pipeline asincrona con più stadi, ad esempio lettura, validazione, trasformazione e persistenza dei dati. È uno scenario molto vicino ai progetti reali e ti aiuta a padroneggiare davvero asyncio.