--- title: "contextvars em Python: estado por requisição em async e threads" url: "https://python.dev.br/blog/python-contextvars-estado-por-requisicao/" markdown_url: "https://python.dev.br/blog/python-contextvars-estado-por-requisicao.MD" description: "Use contextvars.ContextVar para guardar request ID, tenant e usuário por requisição em Python, sem variáveis globais, com exemplos em FastAPI, asyncio, threads e testes." date: "2026-09-18" author: "Equipe Python Dev BR" --- # contextvars em Python: estado por requisição em async e threads Use contextvars.ContextVar para guardar request ID, tenant e usuário por requisição em Python, sem variáveis globais, com exemplos em FastAPI, asyncio, threads e testes. 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](/blog/python-async-await/) usa para isolar estado por task, funciona junto com [FastAPI](/blog/apis-rest-com-fastapi/) e também com [threads](/blog/python-threading-threadpoolexecutor/), 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: ```python # 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 ```python 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](/blog/context-managers-python-with/) 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: ```python 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. ```python 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: ```python 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](/blog/fastapi-background-tasks-celery-redis-2026/) — 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 | Critério | Global do módulo | `threading.local` | `contextvars.ContextVar` | |---|---|---|---| | Isolamento | Nenhum — um valor por processo | Por thread | Por contexto (task async, thread, chamada) | | Seguro com asyncio | Não | Não (mesma thread) | Sim — cada task tem cópia | | Valor default | Sim (atribuição) | Precisa de `try/except` ou init | Sim, no construtor | | Restaurar valor anterior | Manual | Manual | `Token` + `reset()` | | Propaga para task filha | N/A | Não | Sim, via cópia no `create_task` | | Indicado para | Constantes e configuração imutável | Código legado só-thread | Qualquer código novo, sync ou async | 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: ```python 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](/blog/fastapi-background-tasks-celery-redis-2026/) — definindo a variável a partir do payload em vez do contexto do processo.