Aplicações Python modernas executam muitas tarefas ao mesmo tempo. Servidores web processam várias requisições, consumidores leem filas em paralelo e rotinas assíncronas alternam de execução a cada await. Nesse cenário, guardar informações de contexto em variáveis globais pode causar erros difíceis de reproduzir: o identificador de uma requisição pode aparecer no log de outra, um usuário pode herdar dados de outro fluxo e uma função profunda pode exigir parâmetros extras apenas para transportar metadados. O módulo contextvars resolve esse problema ao oferecer variáveis de contexto isoladas para cada fluxo de execução.
Neste guia, você aprenderá como usar ContextVar, tokens, valores padrão e integração com asyncio. Também veremos um padrão prático para correlation IDs, cuidados com threads e testes. Para ampliar seus estudos, consulte também nossos artigos sobre logging estruturado com structlog, configurações seguras com Pydantic Settings, leitura de TOML com tomllib e polimorfismo com singledispatch.
O problema das variáveis globais
Uma variável global funciona quando existe apenas um fluxo linear. Porém, em um servidor assíncrono, várias corrotinas compartilham o mesmo processo. Imagine uma variável global chamada request_id. A requisição A define o valor como A123, realiza uma operação de rede e libera o loop de eventos. Antes de A continuar, a requisição B define a mesma variável como B456. Quando A retoma, ela pode registrar o identificador de B. O código parece correto em testes simples, mas falha sob concorrência.
Passar o identificador como parâmetro em todas as funções evita o compartilhamento, mas cria acoplamento. Funções que não usam diretamente o valor precisam recebê-lo apenas para repassá-lo. Bibliotecas de logging, métricas e auditoria ficam mais difíceis de integrar. contextvars mantém a informação disponível no contexto atual sem transformá-la em estado global compartilhado.
Criando sua primeira ContextVar
from contextvars import ContextVar
request_id: ContextVar[str] = ContextVar("request_id", default="sem-id")
print(request_id.get())
request_id.set("abc-123")
print(request_id.get())O construtor recebe um nome usado principalmente para depuração. O argumento default é opcional, mas costuma ser útil para evitar exceções quando o valor ainda não foi definido. Sem valor padrão, chamar get() antes de set() gera LookupError. Em aplicações críticas, essa exceção pode ser desejável, pois revela que o contexto obrigatório não foi inicializado.
Por que guardar o token
O método set() retorna um token que representa o estado anterior. Use esse token para restaurar o contexto com reset(). Essa prática é essencial quando uma função define um valor temporário.
token = request_id.set("req-789")
try:
executar_operacao()
finally:
request_id.reset(token)O bloco finally garante a restauração mesmo quando ocorre uma exceção. Evite simplesmente definir outro valor no final, porque você pode perder um contexto anterior criado por uma camada superior. O token preserva exatamente o estado que existia antes da alteração.
ContextVar com asyncio
O principal benefício aparece em código assíncrono. Cada tarefa criada pelo asyncio mantém sua própria cópia lógica do contexto. O exemplo a seguir executa duas tarefas concorrentes, mas cada uma lê o identificador correto.
import asyncio
from contextvars import ContextVar
current_job = ContextVar("current_job")
async def etapa():
await asyncio.sleep(0.01)
print(current_job.get())
async def executar(nome: str):
token = current_job.set(nome)
try:
await etapa()
finally:
current_job.reset(token)
async def main():
await asyncio.gather(
executar("importacao"),
executar("relatorio"),
)
asyncio.run(main())Mesmo que as corrotinas alternem de execução, o contexto não é confundido. Essa característica é documentada oficialmente em contextvars na documentação do Python e funciona em conjunto com o modelo de tarefas descrito na documentação de tarefas do asyncio.
Correlation ID em logs
Um caso real é associar todos os logs de uma requisição ao mesmo correlation ID. A camada de entrada cria ou recebe o identificador, armazena-o em uma ContextVar e qualquer função interna pode consultá-lo. Assim, o serviço de banco de dados, o cliente HTTP e o processador de regras não precisam receber o ID como argumento.
from contextvars import ContextVar
import logging
import uuid
correlation_id = ContextVar("correlation_id", default="background")
logger = logging.getLogger(__name__)
async def processar_requisicao():
token = correlation_id.set(str(uuid.uuid4()))
try:
logger.info("inicio", extra={"correlation_id": correlation_id.get()})
await executar_regra()
finally:
correlation_id.reset(token)Se você usa structlog, pode criar um processador que lê a variável e adiciona o campo automaticamente. Isso centraliza o comportamento e evita esquecer o identificador em mensagens específicas. Também é possível guardar tenant, usuário técnico, locale ou trace ID, desde que os dados sejam pequenos e façam sentido durante todo o fluxo.
Não use contexto como depósito de dados
ContextVar não substitui objetos de domínio, parâmetros explícitos ou armazenamento persistente. Evite guardar conexões, grandes estruturas, respostas completas ou informações que precisam atravessar processos. O contexto deve conter metadados leves relacionados à execução atual. Quanto mais valores invisíveis uma função utiliza, mais difícil fica entender suas dependências.
Uma regra prática é usar parâmetros para dados necessários à lógica de negócio e contexto para informações transversais, como rastreamento, auditoria e localização. Se uma função produz resultados diferentes por causa de um valor contextual importante, considere tornar essa dependência explícita.
Threads e execução em executores
Cada thread possui sua própria pilha de contextos. Ao criar tarefas com asyncio.create_task(), o contexto normalmente é copiado. Entretanto, ao enviar trabalho para uma thread manualmente, você deve verificar como o contexto será propagado. A função copy_context() permite capturar o contexto atual e executá-lo em outro local.
from contextvars import copy_context
from concurrent.futures import ThreadPoolExecutor
ctx = copy_context()
with ThreadPoolExecutor() as executor:
futuro = executor.submit(ctx.run, funcao_bloqueante)
futuro.result()Em versões modernas do Python, algumas APIs de alto nível fazem parte dessa propagação automaticamente, mas não presuma esse comportamento para bibliotecas externas. Crie um teste pequeno quando misturar asyncio, threads e executores.
Testando código com ContextVar
Testes devem restaurar o contexto após cada cenário. Um fixture pode definir o valor, entregar o controle ao teste e executar reset() no encerramento. Isso impede que um teste influencie o seguinte.
import pytest
@pytest.fixture
def contexto_teste():
token = request_id.set("teste-001")
yield
request_id.reset(token)
def test_servico(contexto_teste):
assert request_id.get() == "teste-001"Também teste concorrência. Execute duas corrotinas com valores diferentes e confirme que cada uma observa apenas o próprio valor. Esse teste protege contra futuras refatorações que introduzam uma variável global ou fechem o contexto no lugar errado.
Erros comuns
O primeiro erro é chamar set() e nunca restaurar o valor. Em processos de longa duração, isso pode vazar contexto para operações posteriores realizadas na mesma tarefa. O segundo é criar a ContextVar dentro de uma função repetidamente. Declare-a no nível do módulo para manter uma referência estável. O terceiro é usar um valor padrão que mascara falhas. Para dados obrigatórios, prefira não definir padrão e trate LookupError durante o desenvolvimento.
Outro problema é confundir isolamento de contexto com segurança. Uma ContextVar ajuda a separar fluxos, mas não criptografa dados nem impede que código no mesmo contexto os leia. Não armazene segredos sem necessidade. Para credenciais e configurações, use mecanismos próprios e aplique o princípio do menor privilégio.
Organização recomendada
Crie um módulo pequeno, por exemplo app_context.py, com as variáveis e funções auxiliares. Exponha operações como get_request_id() e um context manager para definição temporária. Essa camada evita espalhar chamadas diretas a set() e reset(), facilita testes e permite trocar a implementação.
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Iterator
_request_id = ContextVar("request_id")
@contextmanager
def request_context(value: str) -> Iterator[None]:
token = _request_id.set(value)
try:
yield
finally:
_request_id.reset(token)
def get_request_id() -> str:
return _request_id.get()Conclusão
O módulo contextvars oferece uma forma segura e eficiente de manter metadados por fluxo em aplicações concorrentes. Ele reduz o uso de globais, evita transportar parâmetros por camadas que não precisam deles e integra muito bem com logging, tracing e servidores assíncronos. A implementação correta depende de três hábitos: declarar as variáveis no módulo, restaurar valores com tokens e manter o contexto pequeno e transversal. Com testes de concorrência e uma API auxiliar clara, você obtém observabilidade melhor sem comprometer o isolamento entre requisições.
