Asynchronous programming

What concurrency means here

An asynchronous server has an event loop: a queue that alternates between requests. When a route async def reaches a await (a network wait, for example), it yields control, and the loop serves another request. When the wait ends, the first continues. There is no parallelism in Python code: there is overlapping waits.

This applies to waits for input and output (network, database, disk). For CPU work (processing an image, calculating), the does not help: the calculation occupies the loop the same way. async If a route

The trap: blocking inside `async def`

calls something async def blocking (like or a synchronous library), it does not yield control, and the entire loop time.sleep stops until that finishes. An example with three routes that "take" 0.2 seconds each, called 5 times at the same time: The result is the central point of the chapter. The route

import asyncio
import time

import httpx2
from fastapi import FastAPI

app = FastAPI()


@app.get("/bloqueante")
async def bloqueante():
    time.sleep(0.2)
    return {"ok": True}


@app.get("/cooperativa")
async def cooperativa():
    await asyncio.sleep(0.2)
    return {"ok": True}


@app.get("/em_thread")
def em_thread():
    time.sleep(0.2)
    return {"ok": True}


async def duracao_de_5_chamadas_simultaneas(caminho: str) -> float:
    transporte = httpx2.ASGITransport(app=app)
    async with httpx2.AsyncClient(transport=transporte, base_url="http://teste") as cliente:
        inicio = time.perf_counter()
        await asyncio.gather(*(cliente.get(caminho) for _ in range(5)))
        return time.perf_counter() - inicio


bloq = asyncio.run(duracao_de_5_chamadas_simultaneas("/bloqueante"))
coop = asyncio.run(duracao_de_5_chamadas_simultaneas("/cooperativa"))
thread = asyncio.run(duracao_de_5_chamadas_simultaneas("/em_thread"))

print("async def com time.sleep: as 5 esperas somaram (5 x 0,2 s):", bloq >= 0.9)
print("async def com await asyncio.sleep: esperas sobrepostas:", coop < 0.5)
print("def comum com time.sleep: esperas sobrepostas em threads:", thread < 0.5)

async def com time.sleep: as 5 esperas somaram (5 x 0,2 s): True
async def com await asyncio.sleep: esperas sobrepostas: True
def comum com time.sleep: esperas sobrepostas em threads: True

serialized the five calls (about 1 second in total): async def with time.sleep blocked the loop . Theand the async def with await common finished in about 0.2 seconds, because the waits overlapped. def FastAPI handles the two differently. A route

`def` or `async def`?

FastAPI handles the two differently. A route def common runs on a thread pool, and then a lock on it only occupies one thread. A async def runs in the main loop, and then any lock on it blocks everything.

There is no "always async def". The rule is to match the route type with the library type it uses.

Calling a blocking operation from within `async def`

Sometimes you are in a route async def and need to call a synchronous function. The asyncio.to_thread executes it in a thread and leaves the loop free:

def consulta_lenta_e_sincrona() -> str:
    time.sleep(0.2)
    return "resultado"


@app.get("/ponte")
async def ponte():
    resultado = await asyncio.to_thread(consulta_lenta_e_sincrona)
    return {"resultado": resultado}


ponte_duracao = asyncio.run(duracao_de_5_chamadas_simultaneas("/ponte"))
print("to_thread libera o laço:", ponte_duracao < 0.5)

to_thread libera o laço: True

How many threads

The thread pool of the routes def has a limit (40 by default in AnyIO, the library that FastAPI depends on). With thousands of simultaneous requests blocked on routes def, the limit shows up. Then the options are more processes (--workers), a real asynchronous driver, or moving the slow work to a task queue.

async in the database

SQLAlchemy 2 has an asynchronous version (create_async_engine, AsyncSession), which requires an asynchronous driver (for example, asyncpg or psycopg in asynchronous mode). I only migrate to it when the concurrency gain justifies the complexity: an API with routes def and synchronous session handles a lot of load, and is easier to test.