Python asyncio: async e await

Il modulo asyncio di Python consente di scrivere codice concorrente in un singolo thread usando le parole chiave async e await. Invece di bloccarsi in attesa di risposte di rete o letture di file, un programma asyncio sospende il task in attesa e passa immediatamente ad altre operazioni — riprendendo quando il risultato è disponibile. Questo lo rende lo strumento giusto per programmi I/O-bound come web scraper, client API e server di chat.

Questo capitolo tratta:

• Cosa sono le funzioni async (coroutine) e come differiscono dalle funzioni normali
• L'event loop e come asyncio pianifica il lavoro
• Come attendere risultati, eseguire task concorrentemente con asyncio.gather e asyncio.create_task
• Gestione di eccezioni e timeout nel codice async
• L'asyncio.Queue per i pattern produttore-consumatore
• Quando usare asyncio e quando ricorrere invece al threading

Perché esiste asyncio

Considera un programma che chiama due API una dopo l'altra:

import time

def fetch(name, delay):
    time.sleep(delay)          # blocks the whole program
    return f'data from {name}'

start = time.perf_counter()
r1 = fetch('API A', 1)
r2 = fetch('API B', 1)
print(f'Done in {time.perf_counter() - start:.1f}s')
# Done in 2.0s

Entrambe le chiamate vengono eseguite in sequenza — 2 secondi in totale, anche se ogni chiamata richiede solo 1 secondo di attesa. Con asyncio il programma mette in pausa fetch('API A', ...) mentre aspetta, avvia immediatamente fetch('API B', ...), e entrambe terminano in circa 1 secondo:

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)   # suspends only this coroutine
    return f'data from {name}'

async def main():
    start = time.perf_counter()
    r1, r2 = await asyncio.gather(fetch('API A', 1), fetch('API B', 1))
    print(f'Done in {time.perf_counter() - start:.1f}s')
    # Done in 1.0s

asyncio.run(main())

Coroutine: async def e await

Una funzione definita con async def è chiamata funzione coroutine. Chiamarla non esegue immediatamente il corpo — restituisce un oggetto coroutine che deve essere guidato dall'event loop.

async def greet(name):
    print(f'Hello, {name}!')

# Calling it returns a coroutine object, nothing is printed yet
coro = greet('World')
print(type(coro))   # <class 'coroutine'>

# Run it properly
import asyncio
asyncio.run(greet('World'))
# Hello, World!

All'interno di una coroutine, await sospende l'esecuzione finché l'awaitable (un'altra coroutine, un Task o un Future) non produce un risultato. L'event loop è libero di eseguire altre coroutine mentre una è sospesa.

import asyncio

async def step_one():
    print('Step 1: start')
    await asyncio.sleep(1)     # suspend for 1 second
    print('Step 1: end')
    return 'result-1'

async def main():
    value = await step_one()   # wait for step_one to finish
    print(value)

asyncio.run(main())
# Step 1: start
# Step 1: end
# result-1

Cosa puoi usare con await

• Un'altra coroutine async def
• Un asyncio.Task (creato con asyncio.create_task)
• Un asyncio.Future
• Qualsiasi oggetto con un metodo __await__

Non puoi usare await al di fuori di una funzione async def.

L'event loop

L'event loop è il pianificatore di asyncio. Mantiene una coda di coroutine e task, esegue ciascuno finché non incontra un await, poi passa all'elemento successivo pronto. In genere esiste un solo event loop per thread.

asyncio.run(coro) è il punto di ingresso standard per i programmi asyncio. Crea un nuovo event loop, esegue la coroutine fornita fino al completamento, chiude il loop e restituisce il risultato:

import asyncio

async def compute():
    await asyncio.sleep(0)   # yield control once
    return 6 * 7

result = asyncio.run(compute())
print(result)   # 42

Per la maggior parte delle applicazioni non è mai necessario gestire il loop direttamente — asyncio.run si occupa della creazione e dello smantellamento.

Eseguire task concorrentemente

asyncio.gather

asyncio.gather(*coroutines) pianifica tutte le coroutine fornite per essere eseguite concorrentemente e restituisce i loro risultati nello stesso ordine:

import asyncio

async def fetch_data(name, delay):
    print(f'Start fetching {name}')
    await asyncio.sleep(delay)
    print(f'Done fetching {name}')
    return f'data from {name}'

async def main():
    results = await asyncio.gather(
        fetch_data('API A', 1),
        fetch_data('API B', 2),
        fetch_data('API C', 1),
    )
    print(results)

asyncio.run(main())
# Start fetching API A
# Start fetching API B
# Start fetching API C
# Done fetching API A
# Done fetching API C
# Done fetching API B
# ['data from API A', 'data from API B', 'data from API C']

Tutte e tre le coroutine partono immediatamente. Il tempo totale trascorso corrisponde alla coroutine più lenta (2 s), non alla somma (4 s).

asyncio.create_task

asyncio.create_task(coro) racchiude una coroutine in un Task e la pianifica per l'esecuzione imminente. A differenza di gather, creare un task lo avvia in background mentre la coroutine corrente continua a girare:

import asyncio

async def background_job(name, delay):
    print(f'{name}: start')
    await asyncio.sleep(delay)
    print(f'{name}: end')
    return f'{name} done'

async def main():
    t1 = asyncio.create_task(background_job('Task A', 1))
    t2 = asyncio.create_task(background_job('Task B', 2))

    # Both tasks are already scheduled; await collects their results
    result1 = await t1
    result2 = await t2
    print(result1, result2)

asyncio.run(main())
# Task A: start
# Task B: start
# Task A: end
# Task B: end
# Task A done Task B done

Usa create_task quando vuoi che un task parta immediatamente e intendi raccogliere il suo risultato (o cancellarlo) in seguito. Usa gather quando vuoi avviare un gruppo fisso di coroutine e attendere che tutte terminino insieme.

Output intercalato

Un modo utile per vedere l'event loop in azione è osservare come i task si alternano:

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f'{name}: {i}')
        await asyncio.sleep(1)
    print(f'{name}: done!')

async def main():
    await asyncio.gather(
        count_down('Task A', 3),
        count_down('Task B', 2),
    )

asyncio.run(main())
# Task A: 3
# Task B: 2
# Task A: 2
# Task B: 1
# Task A: 1
# Task B: done!
# Task A: done!

Entrambi i task condividono un solo thread; l'event loop alterna tra di loro ad ogni await asyncio.sleep(1).

Gestione delle eccezioni

Le eccezioni sollevate all'interno di una coroutine si propagano attraverso await proprio come nel codice sincrono. Usa un normale blocco try/except:

import asyncio

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError('something went wrong')

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f'Caught: {e}')

asyncio.run(main())
# Caught: something went wrong

Quando si usa asyncio.gather, se una coroutine solleva un'eccezione le altre non vengono cancellate di default, ma l'eccezione viene rilanciata quando si fa await sulla chiamata gather. Passa return_exceptions=True per raccogliere le eccezioni come valori di ritorno invece:

import asyncio

async def good():
    return 'ok'

async def bad():
    raise RuntimeError('oops')

async def main():
    results = await asyncio.gather(good(), bad(), return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f'Error: {r}')
        else:
            print(f'Result: {r}')

asyncio.run(main())
# Result: ok
# Error: oops

Timeout con asyncio.wait_for

asyncio.wait_for(coro, timeout) esegue una coroutine e la cancella se non termina entro il numero di secondi specificato, sollevando asyncio.TimeoutError:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return 42

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print('Timed out — operation cancelled')

asyncio.run(main())
# Timed out — operation cancelled

Questo è importante per il codice di rete in produzione, dove un server bloccato altrimenti bloccherebbe un task a tempo indeterminato.

asyncio.Queue per i pattern produttore-consumatore

asyncio.Queue è una coda thread-safe e async-aware. È ideale per disaccoppiare i produttori (codice che genera lavoro) dai consumatori (codice che lo elabora):

import asyncio

async def producer(queue):
    for i in range(1, 4):
        print(f'Produced item {i}')
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)   # sentinel to signal consumers to stop

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed item {item}')

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
# Produced item 1
# Consumed item 1
# Produced item 2
# Consumed item 2
# Produced item 3
# Consumed item 3

Per più consumatori, usa queue.task_done() e queue.join() per sapere quando tutti gli elementi sono stati elaborati.

asyncio vs threading

Sia asyncio che il modulo threading di Python consentono di procedere con il lavoro in modo concorrente, ma lo fanno in modo diverso:

Usa asyncio quando controlli il codice I/O e puoi usare librerie compatibili con async (ad es. aiohttp, asyncpg). Usa threading quando dipendi da librerie bloccanti di terze parti che non possono essere rese async.

Per un vero parallelismo CPU, ricorri invece a multiprocessing o concurrent.futures.ProcessPoolExecutor.

Errori comuni

Dimenticare await: Chiamare una funzione async senza await restituisce un oggetto coroutine e non fa nulla. Python emette un RuntimeWarning: coroutine '...' was never awaited per aiutare a individuare questo errore.

async def main():
    asyncio.sleep(1)   # BUG: returns a coroutine, does not sleep
    await asyncio.sleep(1)   # correct

Bloccare l'event loop: Eseguire codice sincrono lento (un ciclo intenso, una chiamata di rete bloccante, time.sleep) all'interno di una coroutine congela l'intero event loop. Avvolgi le chiamate bloccanti con asyncio.to_thread (Python 3.9+) per eseguirle in un thread pool senza bloccare:

import asyncio
import time

def blocking_task():
    time.sleep(2)   # simulates a slow blocking operation
    return 'done'

async def main():
    result = await asyncio.to_thread(blocking_task)
    print(result)

asyncio.run(main())
# done

Usare asyncio.run all'interno di un loop già in esecuzione: I notebook Jupyter eseguono già un event loop. Usa direttamente await coro nelle celle del notebook, o installa nest_asyncio per consentire loop annidati.

Riepilogo di riferimento rapido

Esercitazione