Programação assíncrona

O que é concorrência aqui

Um servidor assíncrono tem um laço de eventos: uma fila que alterna entre as requisições. Quando uma rota async def chega a um await (uma espera de rede, por exemplo), ela cede o controle, e o laço atende outra requisição. Quando a espera termina, a primeira continua. Não há paralelismo de código Python: há sobreposição de esperas.

Isso vale para esperas de entrada e saída (rede, banco, disco). Para trabalho de CPU (processar uma imagem, calcular), o async não ajuda: o cálculo ocupa o laço do mesmo jeito.

A armadilha: bloquear dentro de `async def`

Se uma rota async def chamar algo bloqueante (como time.sleep ou uma biblioteca síncrona), ela não cede o controle, e o laço inteiro para até aquilo terminar. Um exemplo com três rotas que "demoram" 0,2 segundo cada, chamadas 5 vezes ao mesmo tempo:

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

O resultado é o ponto central do capítulo. A rota async def com time.sleep serializou as cinco chamadas (cerca de 1 segundo no total): bloqueou o laço. A async def com await e a def comum terminaram em cerca de 0,2 segundo, porque as esperas se sobrepuseram.

`def` ou `async def`?

O FastAPI trata os dois de forma diferente. Uma rota def comum roda em uma pool de threads, e então um bloqueio nela só ocupa uma thread. Uma async def roda no próprio laço, e então qualquer bloqueio nela para todos.

Não existe "sempre async def". A regra é combinar o tipo da rota com o tipo da biblioteca que ela usa.

Chamar algo bloqueante de dentro de `async def`

Às vezes você está em uma rota async def e precisa chamar uma função síncrona. O asyncio.to_thread a executa em uma thread e deixa o laço livre:

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

Quantas threads

A pool de threads das rotas def tem um limite (40 por padrão no AnyIO, a biblioteca de que o FastAPI depende). Com milhares de requisições simultâneas bloqueadas em rotas def, o limite aparece. Aí as opções são mais processos (--workers), um driver assíncrono de verdade ou mover o trabalho lento para uma fila de tarefas.

async no banco de dados

O SQLAlchemy 2 tem uma versão assíncrona (create_async_engine, AsyncSession), que exige um driver assíncrono (por exemplo, asyncpg ou psycopg em modo assíncrono). Eu só migro para ela quando o ganho de concorrência justifica a complexidade: uma API com rotas def e sessão síncrona atende muita carga, e é mais fácil de testar.