02 - Async/Await in Python: Capire l'Event Loop, Coroutine e Concorrenza I/O-Bound

Mettere async def davanti a una rotta FastAPI non la rende più veloce: la rende capace di aspettare senza occupare un thread. Se poi dentro quella rotta chiami una funzione che blocca (un time.sleep, un client HTTP sincrono, un driver di database tradizionale), fermi l'intero server finché non finisce. Ho misurato la differenza su cento richieste che simulano ciascuna cinquanta millisecondi di attesa, in nove esecuzioni consecutive: 0,09-0,19 s con async def e await asyncio.sleep, 0,18-0,29 s con un def normale, 5,09-5,18 s con async def e time.sleep. Gli intervalli sono larghi perché altri processi occupavano la macchina; le proporzioni no. Il resto dell'articolo spiega perché sono queste.

I listati sono stati eseguiti il 2026-11-12 con CPython 3.12.13, FastAPI 0.136.3 e httpx 0.28.1, su una macchina a 12 core; la riga che comincia per # Output: è l'uscita reale. I tempi cambiano da una macchina e da un'esecuzione all'altra, quindi per ciò che dipende dalla macchina riporto un ordine di grandezza o un intervallo, non un valore «atteso». Nessun server è stato avviato: le richieste ad FastAPI passano per il trasporto ASGI di httpx, nello stesso processo e senza rete. Non è un benchmark di produzione, è un esperimento che isola la variabile che ci interessa.

Il GIL non è il problema che pensi

Il Global Interpreter Lock lascia eseguire il bytecode Python a un solo thread alla volta (è la definizione del glossario della documentazione ufficiale). Se ne conclude spesso che Python «non può fare più cose insieme». È falso per il lavoro che consiste nell'attendere. Un thread che dorme o aspetta una risposta dal database non sta eseguendo bytecode, e un altro thread può andare avanti. Quello che il GIL impedisce è che due thread facciano calcoli nello stesso istante, e questo lo misureremo nell'ultima sezione.

Un'API passa la maggior parte del tempo ad aspettare: il database, un altro servizio, un file. Immagina un cameriere che prende l'ordine a un tavolo, lo porta in cucina e, invece di restare lì finché il piatto è pronto, va al tavolo successivo; torna in cucina quando lo chiamano. È questo che fa asyncio: un solo thread, un solo cameriere, e la regola che ogni attesa lunga deve essere dichiarata con await per consentire il cambio di tavolo. Python 3.13 ha introdotto in via sperimentale una build senza GIL (lo dicono le note di rilascio della versione 3.13), ma tutto ciò che segue vale per l'interprete normale.

Event loop e coroutine, con un tempo da leggere

L'event loop è il ciclo che tiene l'elenco delle coroutine pronte, ne esegue una finché non incontra un await su un'attesa vera, la mette da parte e passa alla successiva. Una coroutine è una funzione dichiarata con async def: chiamarla non la esegue, produce un oggetto che il loop può far avanzare. Il punto che sfugge a chi comincia è che await da solo non crea concorrenza: sospende la coroutine corrente finché l'attesa è finita. La concorrenza nasce quando più coroutine sono in attesa nello stesso momento.

Il listato confronta due modi di leggere tre fonti con latenze di 0,2, 0,3 e 0,1 secondi (il caso tipico di una pagina che deve consultare utenti, ordini e preferenze). Nella versione in sequenza ogni await aspetta il precedente. In quella con asyncio.gather le tre coroutine partono insieme e il tempo totale è quello della più lenta.

# sequenza.py
import asyncio
import time


async def leggi(nome: str, attesa: float) -> str:
    await asyncio.sleep(attesa)          # qui la coroutine cede il controllo al loop
    return f"{nome} ({attesa} s)"


async def in_sequenza():
    utenti = await leggi("utenti", 0.2)
    ordini = await leggi("ordini", 0.3)
    preferenze = await leggi("preferenze", 0.1)
    return [utenti, ordini, preferenze]


async def in_concorrenza():
    return await asyncio.gather(
        leggi("utenti", 0.2),
        leggi("ordini", 0.3),
        leggi("preferenze", 0.1),
    )


async def misura(funzione):
    inizio = time.perf_counter()
    risultato = await funzione()
    print(f"{funzione.__name__:14s} {time.perf_counter() - inizio:.2f} s  {risultato}")


async def main():
    await misura(in_sequenza)
    await misura(in_concorrenza)


asyncio.run(main())

# Output:
# in_sequenza    0.60 s  ['utenti (0.2 s)', 'ordini (0.3 s)', 'preferenze (0.1 s)']
# in_concorrenza 0.30 s  ['utenti (0.2 s)', 'ordini (0.3 s)', 'preferenze (0.1 s)']

Il risultato è 0,60 contro 0,30 secondi: la somma delle attese contro il massimo. I tempi di questo esperimento sono quasi esatti perché le attese sono artificiali; con un database vero avranno una variabilità tutta loro. Nota che il codice della funzione leggi è identico nei due casi: a cambiare è solo il modo in cui la chiami. Se le tre letture dipendessero l'una dall'altra (gli ordini hanno bisogno dell'id dell'utente), la sequenza sarebbe obbligata e gather non servirebbe.

La trappola più comune: la coroutine che non parte

Dimenticare l'await non produce un errore: produce un oggetto. La riga risultato = leggi("utenti") non ha eseguito nulla, e se il tipo restituito non lo guardi nessuno ti avvisa finché l'oggetto non viene eliminato. Python lo segnala con un RuntimeWarning, che finisce su stderr in mezzo agli altri messaggi e spesso passa inosservato. Il listato cattura l'avviso per mostrartelo.

# trappola.py
import asyncio
import gc
import warnings


async def leggi(nome: str) -> str:
    await asyncio.sleep(0.1)
    return nome


with warnings.catch_warnings(record=True) as avvisi:
    warnings.simplefilter("always")
    risultato = leggi("utenti")      # manca l'await: la funzione NON è stata eseguita
    print(type(risultato).__name__)
    del risultato
    gc.collect()

print(avvisi[0].category.__name__, "-", avvisi[0].message)
print(asyncio.run(leggi("utenti")))   # così sì: il loop la esegue

# Output:
# coroutine
# RuntimeWarning - coroutine 'leggi' was never awaited
# utenti

Il tipo è coroutine, e il messaggio è «coroutine 'leggi' was never awaited». È la firma di un await mancato. L'ultima riga mostra la chiamata corretta: la coroutine deve essere consegnata a un loop, con await dentro un'altra coroutine oppure con asyncio.run al livello più alto. In FastAPI il caso tipico è dimenticare await su una chiamata a un client async dentro la rotta: la risposta non contiene i dati che ti aspettavi, perché hai in mano la coroutine e non il suo risultato.

gather, create_task e TaskGroup: che cosa succede quando uno fallisce

Ci sono tre modi per far girare più coroutine insieme, e la differenza si vede soprattutto quando una di esse fallisce:

• asyncio.gather(...): aspetta tutte e restituisce la lista dei risultati nell'ordine in cui le hai passate.
• asyncio.create_task(...): avvia subito una coroutine in background e ti restituisce un oggetto Task da aspettare più tardi.
• asyncio.TaskGroup (Python 3.11 e successivi): un blocco async with che aspetta tutti i task creati al suo interno e li ferma se uno fallisce.

Il criterio per scegliere è la gestione degli errori. Con gather vuoi i risultati parziali, con TaskGroup vuoi tutto o niente, con create_task hai un lavoro che deve vivere più a lungo della funzione che lo avvia, e allora devi tenere un riferimento al task: la documentazione avverte che il loop conserva solo riferimenti deboli, quindi un task senza riferimenti può essere eliminato dal garbage collector prima di finire.

# errori.py
import asyncio


async def lavoro(nome: str, attesa: float, fallisce: bool = False) -> str:
    await asyncio.sleep(attesa)
    if fallisce:
        raise ValueError(f"{nome} ha fallito")
    print(f"  {nome} completato")
    return nome


async def con_gather():
    print("gather:")
    try:
        await asyncio.gather(lavoro("A", 0.1, fallisce=True), lavoro("B", 0.3))
    except ValueError as e:
        print("  sollevata:", e)
    await asyncio.sleep(0.4)             # tempo perché B finisca, se non è stato fermato


async def con_return_exceptions():
    print("gather con return_exceptions=True:")
    risultati = await asyncio.gather(
        lavoro("A", 0.1, fallisce=True), lavoro("B", 0.3), return_exceptions=True
    )
    print("  ", risultati)


async def con_taskgroup():
    print("TaskGroup:")
    try:
        async with asyncio.TaskGroup() as gruppo:
            gruppo.create_task(lavoro("A", 0.1, fallisce=True))
            gruppo.create_task(lavoro("B", 0.3))
    except* ValueError as gruppo_errori:
        print("  sollevate:", [str(e) for e in gruppo_errori.exceptions])
    await asyncio.sleep(0.4)


async def main():
    await con_gather()
    await con_return_exceptions()
    await con_taskgroup()


asyncio.run(main())

# Output:
# gather:
#   sollevata: A ha fallito
#   B completato
# gather con return_exceptions=True:
#   B completato
#    [ValueError('A ha fallito'), 'B']
# TaskGroup:
#   sollevate: ['A ha fallito']

Tre comportamenti, tre risultati. Con gather senza opzioni l'eccezione di A arriva subito a chi aspetta, ma B non viene fermato: continua e completa, come dice la documentazione («Other awaitables … won't be cancelled»). Se non ti aspettavi che B finisse, hai un lavoro che gira senza che nessuno ne raccolga il risultato. Con return_exceptions=True nessuna eccezione viene sollevata: la lista contiene l'errore al posto del risultato di A, e sta a te controllarlo con isinstance. Con TaskGroup, invece, B non completa: il gruppo lo cancella appena A fallisce, e le eccezioni escono raggruppate in un ExceptionGroup, che si cattura con la sintassi except*. È il comportamento da preferire quando i lavori del gruppo hanno senso solo insieme; per i dettagli di cancellazione c'è async/await avanzato: TaskGroup e cancellation.

async def o def in una rotta FastAPI

FastAPI accetta entrambe, e fa cose diverse. Una rotta async def viene eseguita direttamente nell'event loop; una rotta def viene eseguita in un thread di un pool (lo dice la pagina Concurrency and async / await), così non blocca il loop. Dunque nessuna delle due «è sbagliata»: sbagliato è metterci dentro l'attesa sbagliata. La documentazione di FastAPI dà tre indicazioni, e le riporto ordinate:

• Se la libreria che usi ha una versione con await (httpx per HTTP, i driver async per i database), dichiara la rotta con async def e usa await.
• Se la libreria non supporta await, come la maggior parte dei client di database tradizionali, dichiara la rotta con un def normale.
• Se non sai che cosa fare, usa def; se la rotta non aspetta nulla, async def va bene anche senza await.

Il filo che unisce le tre regole è «non mettere una chiamata bloccante dentro l'event loop». Un def che chiama una libreria sincrona è corretto, perché il blocco avviene in un thread e il loop resta libero; un async def che chiama la stessa libreria è l'errore, perché il blocco avviene nel loop e fa aspettare tutte le altre richieste. L'esperimento che segue mette le tre combinazioni una accanto all'altra.

# bench.py
import asyncio
import time

import httpx
from anyio import to_thread
from fastapi import FastAPI

app = FastAPI()
ATTESA = 0.05          # secondi di "attesa del database" simulata


@app.get("/sync")
def sync_bloccante():                 # def: FastAPI la esegue in un thread del pool
    time.sleep(ATTESA)
    return {"ok": True}


@app.get("/async")
async def async_non_bloccante():      # async def: l'attesa cede il controllo al loop
    await asyncio.sleep(ATTESA)
    return {"ok": True}


@app.get("/async-errato")
async def async_bloccante():          # async def + chiamata bloccante: il caso da evitare
    time.sleep(ATTESA)
    return {"ok": True}


async def misura(etichetta: str, percorso: str, n: int = 100):
    trasporto = httpx.ASGITransport(app=app)       # nessuna rete: l'app gira nel processo
    async with httpx.AsyncClient(transport=trasporto, base_url="http://test") as client:
        inizio = time.perf_counter()
        risposte = await asyncio.gather(*(client.get(percorso) for _ in range(n)))
        durata = time.perf_counter() - inizio
    ok = sum(r.status_code == 200 for r in risposte)
    print(f"{etichetta:24s} {n} richieste in {durata:5.2f} s  (ok: {ok})")


async def main():
    limite = to_thread.current_default_thread_limiter()
    print("thread del pool di default:", limite.total_tokens)
    await misura("/async", "/async")
    await misura("/sync", "/sync")
    limite.total_tokens = 100
    await misura("/sync con pool da 100", "/sync")
    limite.total_tokens = 40
    await misura("/async-errato", "/async-errato")


asyncio.run(main())

# Output:
# thread del pool di default: 40
# /async                   100 richieste in  0.11 s  (ok: 100)
# /sync                    100 richieste in  0.19 s  (ok: 100)
# /sync con pool da 100    100 richieste in  0.10 s  (ok: 100)
# /async-errato            100 richieste in  5.13 s  (ok: 100)

Le tre righe sono la tesi dell'articolo. Con /async le cento attese da 50 ms si sovrappongono tutte: da 0,09 a 0,19 s nelle nove esecuzioni (nell'output qui sopra, una di esse). Con /sync FastAPI usa il pool di thread, che per default ha 40 posti (il limite di AnyIO, riportato anche nella prima riga dell'output, e descritto nella documentazione di AnyIO): cento richieste in tre ondate, 3 × 0,05 s, cioè 0,15 s ideali; misurati da 0,18 a 0,29 s, con l'avvio dei thread e il carico della macchina a fare il resto. Portando il limite a cento (quarta riga) il tempo si riavvicina a quello dell'async. Con /async-errato ogni richiesta ferma il loop per 50 ms, nessuna si sovrappone alle altre, e il totale è la somma: 100 × 0,05 = 5 secondi, misurati da 5,09 a 5,18. Questo è l'unico dei tre valori che non dipende dal carico della macchina, perché è dominato dalla somma delle attese.

Due avvertenze sull'esperimento. La prima: la differenza fra /async e /sync qui è piccola perché quaranta thread che dormono costano poco; con molte migliaia di connessioni contemporanee il pool da 40 diventa il tetto e la differenza cresce, ma non l'ho misurata e non do numeri. La seconda: time.sleep è il caso di scuola, ma lo stesso blocco lo producono requests.get o una query con psycopg2 dentro una rotta async def. Se il sintomo è «il server risponde a una richiesta per volta anche con un solo processo libero», cerca lì.

Il lavoro di CPU: né await né thread

Async non aiuta quando il tempo si spende a calcolare. Un ciclo di calcolo non ha punti in cui cedere il controllo: se sta nel loop lo occupa fino alla fine, e se sta in un thread si scontra con il GIL. L'unico modo per usare più core con CPython normale è usare più processi. Il listato misura tre modi di eseguire quattro calcoli identici da circa 0,15-0,2 s l'uno: direttamente nel loop, in thread con asyncio.to_thread, in processi con ProcessPoolExecutor. In parallelo gira una coroutine «sentinella» che dovrebbe svegliarsi ogni 10 ms: la sua pausa più lunga misura per quanto il loop è rimasto occupato.

# cpu.py
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

N = 3_000_000
LAVORI = 4          # quattro calcoli da fare "insieme"


def calcolo(n: int) -> int:
    return sum(i * i for i in range(n))     # lavoro puro di CPU, nessuna attesa


async def calcoli_nel_loop():
    return [calcolo(N) for _ in range(LAVORI)]      # chiamate dirette: il loop resta occupato


async def battito(gap: list[float], fermo: asyncio.Event):
    ultimo = time.perf_counter()
    while not fermo.is_set():
        await asyncio.sleep(0.01)           # dovrebbe tornare ogni ~10 ms
        ora = time.perf_counter()
        gap.append(ora - ultimo)
        ultimo = ora


async def prova(nome: str, esegui):
    gap: list[float] = []
    fermo = asyncio.Event()
    sentinella = asyncio.create_task(battito(gap, fermo))
    await asyncio.sleep(0.05)
    inizio = time.perf_counter()
    await esegui()
    durata = time.perf_counter() - inizio
    fermo.set()
    await sentinella
    print(f"{nome:22s} durata {durata:.2f} s, pausa massima del loop {max(gap) * 1000:4.0f} ms")


async def main():
    loop = asyncio.get_running_loop()
    await prova("direttamente nel loop", calcoli_nel_loop)
    await prova(
        "asyncio.to_thread",
        lambda: asyncio.gather(*(asyncio.to_thread(calcolo, N) for _ in range(LAVORI))),
    )
    with ProcessPoolExecutor(max_workers=LAVORI) as pool:
        await prova(
            "ProcessPoolExecutor",
            lambda: asyncio.gather(*(loop.run_in_executor(pool, calcolo, N) for _ in range(LAVORI))),
        )


if __name__ == "__main__":
    asyncio.run(main())

# Output:
# direttamente nel loop  durata 0.61 s, pausa massima del loop  618 ms
# asyncio.to_thread      durata 0.66 s, pausa massima del loop  139 ms
# ProcessPoolExecutor    durata 0.28 s, pausa massima del loop   19 ms

L'output del listato è una delle nove esecuzioni. Leggi le due colonne separatamente. La durata: nel loop e nei thread è praticamente la stessa (il GIL non permette ai quattro thread di calcolare insieme), con i processi scende in modo netto perché i calcoli girano su core diversi. In nove esecuzioni consecutive ho misurato, per i tre modi, durate da 0,56 a 1,0 s, da 0,59 a 1,26 s e da 0,16 a 0,52 s. La pausa del loop: nel loop è quasi uguale alla durata (il server non risponde a nessuno per tutto quel tempo), nei thread scende a un valore compreso fra circa 105 e 400 ms, con i processi resta fra i 15 e i 32 ms. Il thread dunque serve a tenere il loop vivo, non ad andare più veloci.

I processi hanno un costo: funzione e argomenti devono essere serializzabili con pickle e la funzione deve essere importabile dai processi figli (lo dice la documentazione di concurrent.futures), da cui il controllo if __name__ == "__main__" nel listato. Per lavori di pochi millisecondi il costo di spostare i dati supera il guadagno. In una rotta FastAPI il pool si crea una volta all'avvio dell'applicazione, non a ogni richiesta, e il lavoro davvero pesante (inferenza di un modello, elaborazione di video) conviene metterlo in una coda di lavori, con la rotta che restituisce subito un identificativo.

Posso usare requests dentro una rotta async def?

No: requests è sincrona, e dentro una coroutine ferma il loop per tutta la durata della chiamata, come nella terza rotta dell'esperimento. Hai due alternative corrette: dichiarare la rotta con def e lasciare che FastAPI la esegua in un thread, oppure sostituire la libreria con httpx.AsyncClient e usare await. Se non puoi sostituirla e la rotta deve restare async, avvolgi la chiamata in await asyncio.to_thread(...).

async def è sempre più veloce di def?

No, e l'esperimento lo mostra: con le attese sovrapposte i due approcci erano nello stesso ordine di grandezza (0,09 s contro 0,18 s), e la differenza si fa sentire solo quando il pool da 40 thread si satura. Il vantaggio di async def è la capacità di tenere molte attese aperte con poche risorse; lo scotto è la disciplina richiesta: un solo await mancato o una sola chiamata bloccante rovina tutto. Per questo la documentazione suggerisce def quando non sei sicuro.

Che cosa sai fare adesso

Sai spiegare perché il GIL non impedisce la concorrenza su attese di rete, distinguere una sequenza di await da un gather e leggerne la differenza in secondi, e riconoscere dal messaggio «was never awaited» un await dimenticato. Sai scegliere fra gather, create_task e TaskGroup in base a che cosa deve succedere quando un lavoro fallisce. E sai decidere fra def e async def in FastAPI con una regola che si verifica con un cronometro: quale libreria chiamo, e se supporta await.

Il compito che resta, prima di mettere il servizio in produzione, è descrivere e validare i dati che entrano ed escono: lo fa Pydantic, ed è l'argomento di Pydantic v2: validazione avanzata. Per il punto di partenza, cioè la struttura del servizio e la documentazione generata, trovi FastAPI da zero; per mettere in produzione un servizio con modelli, FastAPI e Uvicorn in produzione.