contextvars em Python: estado por requisição em async e threads

Um problema clássico em APIs Python: você quer que cada log, métrica e consulta ao banco carregue o request ID, o tenant ou o usuário logado da requisição atual — sem empilhar esses argumentos em todas as funções. A resposta curta é: use contextvars.ContextVar da biblioteca padrão. É o mecanismo que o próprio asyncio usa para isolar estado por task, funciona junto com FastAPI e também com threads, e substitui tanto variáveis globais quanto threading.local.

Este guia mostra o problema do estado compartilhado, o uso básico de ContextVar com get, set e Token, o padrão de middleware de request ID em FastAPI, como propagar contexto para threads e ThreadPoolExecutor, e como testar tudo isso com pytest.

O problema: global + asyncio vaza estado entre requisições

A tentação é guardar em uma variável global:

# ANTI-PADRÃO em código async
current_user_id: int | None = None

async def middleware(request, call_next):
    global current_user_id
    current_user_id = extrair_user(request)  # compartilhado por TODO o processo
    return await call_next(request)

Em um servidor ASGI, todas as requisições rodam na mesma thread, intercaladas pelo event loop. Entre o set de uma requisição e o uso em outra, o valor já pode ter sido sobrescrito: o log da requisição A sai com o usuário da requisição B. threading.local não resolve — async não troca de thread, então o estado continua compartilhado.

É exatamente isso que contextvars corrige: cada task asyncio executa com uma cópia do contexto, e um set só altera a cópia da task atual.

ContextVar: get, set e Token

from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar("request_id", default="-")


def log(msg: str) -> None:
    print(f"[{request_id.get()}] {msg}")


log("início")          # [-] início — usa o default
token = request_id.set("req-123")
log("processando")     # [req-123] processando
request_id.reset(token)
log("fim")             # [-] fim — voltou ao valor anterior

Três pontos importantes:

• ContextVar deve ser criada no nível do módulo (topo do arquivo), uma única vez — como uma global de leitura. O que varia por contexto é o valor, não a variável.
• set() retorna um Token; reset(token) restaura o valor anterior. Dentro de um context manager ou middleware síncrono, sempre faça set e reset em par try/finally.
• default evita LookupError quando o valor nunca foi definido — útil em scripts, Celery workers e testes que não passam pelo middleware.

Por que asyncio isola automaticamente

Quando o event loop cria uma task com asyncio.create_task(), ele chama implicitamente contextvars.copy_context() e roda a corrotina naquela cópia. Consequências práticas:

import asyncio
from contextvars import ContextVar

valor: ContextVar[str] = ContextVar("valor", default="-")


async def filho(nome: str) -> None:
    await asyncio.sleep(0.01)  # simula I/O intercalado
    print(nome, "->", valor.get())


async def main() -> None:
    valor.set("A")
    t1 = asyncio.create_task(filho("task1"))   # cópia com valor="A"
    valor.set("B")
    t2 = asyncio.create_task(filho("task2"))   # cópia com valor="B"
    await asyncio.gather(t1, t2)

asyncio.run(main())
# task1 -> A
# task2 -> B

Cada task vê o valor do momento da criação. Um set dentro da task não afeta nem a task-mãe nem as irmãs. É esse comportamento que faz o padrão funcionar em servidores: cada requisição HTTP roda em sua própria task (ou grupo de tasks), com contexto isolado.

Padrão FastAPI: middleware de request ID

Combinação comum em produção: gerar um ID por requisição, devolvê-lo no header X-Request-ID e injetá-lo em todos os logs.

import logging
import uuid
from contextvars import ContextVar

from fastapi import FastAPI, Request

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("app")


async def request_id_middleware(request: Request, call_next):
    rid = request.headers.get("x-request-id") or uuid.uuid4().hex[:12]
    request_id_var.set(rid)
    try:
        response = await call_next(request)
    except Exception:
        logger.exception("falha na requisição")
        raise
    response.headers["X-Request-ID"] = rid
    return response


app = FastAPI()
app.middleware("http")(request_id_middleware)


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True


logger.addFilter(ContextFilter())


@app.get("/pagar")
async def pagar():
    logger.info("processando pagamento")  # INFO [app] [a1b2c3d4e5f6] processando pagamento
    return {"status": "ok"}

Refinamentos que valem em produção:

• Use o formato %(request_id)s no formatter do logging em vez de recompor a mensagem.
• Some X-Request-ID à resposta de erro, para o suporte correlacionar o bug reportado.
• Se a requisição chega de um gateway que já gera o ID, reaproveite-o (como acima) em vez de criar outro — os dois lados do trace ficam amarrados.

O mesmo padrão serve para tenant em APIs multi-tenant, usuário autenticado em logs de auditoria e idioma/localização por requisição.

Threads e ThreadPoolExecutor: propague com copy_context

Corrotinas copiam o contexto; threads não. Se você usa run_in_executor ou um pool de threads para chamar bibliotecas síncronas (boto3, pandas, drivers de banco), o contexto precisa ir junto explicitamente:

import asyncio
import contextvars
from concurrent.futures import ThreadPoolExecutor

tenant: ContextVar[str] = ContextVar("tenant", default="-")

executor = ThreadPoolExecutor(max_workers=4)


def consulta_sincrona(sql: str) -> str:
    # roda em outra thread, mas vê o contexto capturado
    return f"tenant={tenant.get()} sql={sql}"


async def handler() -> str:
    ctx = contextvars.copy_context()          # captura ANTES de submeter
    loop = asyncio.get_running_loop()
    resultado = await loop.run_in_executor(
        executor, lambda: ctx.run(consulta_sincrona, "SELECT 1")
    )
    return resultado


async def main() -> None:
    tenant.set("acme")
    print(await handler())   # tenant=acme sql=SELECT 1


asyncio.run(main())

Cuidados:

• Capture o contexto no momento da submissão, fora da thread — capturar dentro dela já é tarde.
• ctx.run() não pode executar o mesmo objeto Context em paralelo em duas threads; capture um contexto por submissão (como no exemplo).
• Contextos também propagam por functools.partial e workers de Celery — mas cada worker tem seu próprio processo, então o padrão lá é definir a variável no início da task a partir do payload da mensagem.

contextvars vs threading.local vs global

Regra prática: use global apenas para valores imutáveis; migre threading.local para ContextVar assim que o código tocar asyncio; e nunca trate contexto como cache de longo prazo — é estado de uma operação, não uma sessão.

Testando com pytest

Como o contexto é isolado por chamada, os testes ficam determinísticos:

import pytest

from app import app, request_id_var


def test_response_carrega_request_id():
    with pytest.MonkeyPatch.context() as mp:
        # garante estado limpo mesmo rodando em paralelo
        token = request_id_var.set("teste-1")
        try:
            client = TestClient(app)
            resp = client.get("/pagar", headers={"x-request-id": "teste-1"})
            assert resp.headers["x-request-id"] == "teste-1"
        finally:
            request_id_var.reset(token)

Dicas:

• Faça set/reset com try/finally (ou um fixture) para não vazar o valor entre testes quando o runner executa tudo na mesma task.
• pytest-asyncio cria tasks por teste, então o isolamento costuma vir de graça — mas o reset explícito documenta a intenção.
• Escreva um teste de concorrência que dispara duas chamadas com IDs diferentes em paralelo e afirma que cada log contém o ID certo; é o teste que pega regressões para globais acidentais.

Conclusão

contextvars resolve o estado por requisição com a biblioteca padrão: crie a ContextVar no topo do módulo, faça set no middleware, leia com get em qualquer camada e use copy_context().run quando cruzar a fronteira para threads. É o mesmo mecanismo interno que asyncio usa, então o custo é baixo e o comportamento é previsível. Se o seu próximo passo é observabilidade completa, combine o request ID com logs estruturados e, para cargas em segundo plano, com o modelo de background tasks e filas — definindo a variável a partir do payload em vez do contexto do processo.