Variáveis de contexto são um mecanismo do Python para armazenar dados que devem ser isolados entre diferentes contextos de execução, como corrotinas, tarefas ou threads. O módulo contextvars, introduzido na versão 3.7, fornece uma API para criar e manipular essas variáveis de forma segura, garantindo que cada contexto tenha seu próprio valor.

Em programação assíncrona, especialmente com asyncio, é comum precisar de informações que pertencem a uma requisição ou tarefa específica, como um ID de usuário, um token de autenticação ou configurações de logging. Variáveis de contexto resolvem esse problema sem precisar passar esses dados explicitamente por todos os parâmetros das funções, tornando o código mais limpo e menos acoplado.

contextvars

O módulo contextvars fornece a classe ContextVar para declarar uma nova variável de contexto. Você cria uma instância no nível do módulo e a utiliza em qualquer lugar do seu código. Cada ContextVar tem um nome (para fins de debug) e um valor padrão opcional.

Para acessar o valor atual, use o método get(). Para definir um novo valor, use o método set(). É importante notar que set() retorna um token, que pode ser usado para reverter a alteração (como um mecanismo de reset). Isso é útil em cenários onde você deseja temporariamente mudar o valor e depois restaurar o anterior.

Exemplo básico:

import contextvars

# Cria uma variável de contexto com valor padrão
user_id = contextvars.ContextVar('user_id', default=None)

# Obtém o valor atual (None)
print(user_id.get())

# Define um novo valor
token = user_id.set(42)
print(user_id.get())  # 42

# Restaura o valor anterior
user_id.reset(token)
print(user_id.get())  # None

O método get() pode receber um valor padrão como argumento, que é usado se a variável não tiver valor definido no contexto atual. Por exemplo: user_id.get('desconhecido').

Além disso, o módulo fornece a classe Context, que representa um conjunto de valores de variáveis de contexto. Você pode copiar um contexto, executar código em um contexto específico usando Context.run(), ou criar um novo contexto vazio com contextvars.Context().

ctx = contextvars.Context()
ctx.run(user_id.set, 99)  # executa a função set no contexto ctx
print(user_id.get())  # None, pois o valor foi definido apenas no ctx

Casos de uso (async)

Em aplicações assíncronas, cada tarefa (Task) ou corrotina (Coroutine) tem seu próprio contexto. Isso significa que, ao usar asyncio.create_task(), a nova tarefa herda uma cópia do contexto atual, mas as alterações feitas dentro dela não afetam o contexto externo. Isso é essencial para isolar dados entre requisições concorrentes.

Um caso comum é ter um middleware em um servidor web (como FastAPI ou aiohttp) que captura o ID do usuário autenticado e o armazena em uma variável de contexto. Depois, qualquer função chamada dentro daquela requisição pode acessar esse ID sem precisar recebê-lo como parâmetro.

Exemplo:

import asyncio
import contextvars

usuario_atual = contextvars.ContextVar('usuario_atual', default='anonimo')

async def processar_requisicao():
    # Simula uma requisição
    usuario_atual.set('joao')
    await asyncio.sleep(0.1)
    print(f'Requisição processada por {usuario_atual.get()}')

async def main():
    # Cria duas tarefas concorrentes
    task1 = asyncio.create_task(processar_requisicao())
    task2 = asyncio.create_task(processar_requisicao())
    await asyncio.gather(task1, task2)

asyncio.run(main())

Neste exemplo, cada tarefa define seu próprio valor para usuario_atual, e as impressões mostram que cada uma mantém seu valor isolado. Isso é fundamental para evitar que dados de uma requisição vazem para outra.

Outro caso é o rastreamento distribuído (tracing), onde você quer propagar um ID de correlação por todas as operações assíncronas de uma mesma requisição, sem passá-lo explicitamente.

vs thread-local

Antes de contextvars, a forma comum de isolamento era usar threading.local(), que armazena dados por thread. Em aplicações com threads, cada thread tem seu próprio espaço de armazenamento. No entanto, em programação assíncrona, as corrotinas são executadas em uma única thread, e o threading.local() não funciona corretamente porque todas as corrotinas compartilham a mesma thread.

Por exemplo, se você usar threading.local() dentro de um loop asyncio, todas as tarefas verão o mesmo valor, pois compartilham a thread. Isso quebra o isolamento esperado.

Com contextvars, o contexto é vinculado à tarefa ou corrotina, não à thread. Isso permite que múltiplas corrotinas concorrentes na mesma thread tenham valores diferentes para a mesma variável.

Além disso, contextvars funciona bem com asyncio, pois o próprio loop de eventos gerencia a troca de contexto quando há mudança de tarefa. Também é possível usar contextvars em código síncrono, mas o benefício é maior em código assíncrono.

Comparação rápida:

  • threading.local(): isolamento por thread; não funciona adequadamente em código async.
  • contextvars: isolamento por contexto (tarefa/corrotina); funciona em código síncrono e assíncrono.

Exemplos

Vamos explorar alguns exemplos práticos de uso de contextvars.

Exemplo 1: Logging com contexto

Você pode incluir informações de contexto em mensagens de log, como o ID da requisição ou o usuário atual.

import contextvars
import logging

request_id = contextvars.ContextVar('request_id', default='-')

logger = logging.getLogger(__name__)

async def handle_request():
    request_id.set('abc123')
    logger.info('Iniciando processamento')
    # ...

# Configurar um filtro para adicionar o request_id ao log
class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get()
        return True

logger.addFilter(RequestIdFilter())

Exemplo 2: Middleware em FastAPI

Um uso comum é em frameworks web como FastAPI, onde você pode capturar o usuário autenticado e armazená-lo em uma variável de contexto.

import contextvars
from fastapi import FastAPI, Request

app = FastAPI()
current_user = contextvars.ContextVar('current_user', default=None)

@app.middleware("http")
async def add_user_to_context(request: Request, call_next):
    # Suponha que o usuário seja obtido de um token ou sessão
    user = await get_user_from_request(request)
    current_user.set(user)
    response = await call_next(request)
    return response

@app.get("/")
async def read_root():
    user = current_user.get()
    return {"message": f"Olá, {user.name}"}

Exemplo 3: Propagação de contexto em tarefas

Quando você cria uma nova tarefa com asyncio.create_task(), ela herda uma cópia do contexto atual. Isso é útil para propagar informações como IDs de correlação.

import asyncio
import contextvars

correlation_id = contextvars.ContextVar('correlation_id', default=None)

async def worker():
    print(f'Worker vê correlation_id: {correlation_id.get()}')

async def main():
    correlation_id.set('corr-123')
    task = asyncio.create_task(worker())
    await task

asyncio.run(main())

Neste caso, o worker verá o valor 'corr-123' porque herdou o contexto da tarefa principal.

Boas práticas

Ao usar contextvars, algumas boas práticas podem ajudar:

  • Defina as variáveis de contexto no nível do módulo para que sejam acessíveis em todo o código.
  • Use valores padrão sensatos para evitar erros quando o valor não foi definido.
  • Evite armazenar objetos mutáveis em variáveis de contexto, pois podem causar efeitos colaterais inesperados.
  • Em bibliotecas, exponha as variáveis de contexto como parte da API pública, mas documente claramente.
  • Use ContextVar.set() com cuidado em código assíncrono, pois o valor é restrito ao contexto atual.

Referências

Exercícios

  1. Crie uma variável de contexto chamada config com valor padrão None. Em seguida, defina um valor para ela e imprima. Depois, redefina para None usando o token.

    ✓ Resposta: 
    import contextvars

config = contextvars.ContextVar('config', default=None)

# Define um valor
token = config.set({'debug': True})
print(config.get())  # {'debug': True}

# Restaura o valor padrão
config.reset(token)
print(config.get())  # None

  2. Escreva um programa assíncrono que use uma variável de contexto para armazenar um número. Crie duas tarefas que definem valores diferentes e imprimem o valor após um pequeno atraso. Verifique que cada tarefa vê seu próprio valor.

    ✓ Resposta: 
    import asyncio
import contextvars

numero = contextvars.ContextVar('numero', default=0)

async def tarefa(n):
    numero.set(n)
    await asyncio.sleep(0.1)
    print(f'Tarefa {n} vê: {numero.get()}')

async def main():
    await asyncio.gather(tarefa(1), tarefa(2))

asyncio.run(main())

  3. Explique por que threading.local() não é adequado para programação assíncrona e como contextvars resolve esse problema.

    ✓ Resposta:

    Em programação assíncrona, todas as corrotinas são executadas em uma única thread (se você usar asyncio). threading.local() armazena dados por thread, então todas as corrotinas compartilhariam o mesmo valor, quebrando o isolamento. contextvars cria um contexto por tarefa/corrotina, permitindo que cada uma tenha seu próprio valor, mesmo na mesma thread.

  4. Implemente um middleware simples para um servidor web (pode ser em FastAPI) que armazena o endereço IP do cliente em uma variável de contexto. Depois, em uma rota, acesse esse valor.

    ✓ Resposta: 
    import contextvars
from fastapi import FastAPI, Request

app = FastAPI()
client_ip = contextvars.ContextVar('client_ip', default='desconhecido')

@app.middleware("http")
async def capture_ip(request: Request, call_next):
    client_ip.set(request.client.host)
    response = await call_next(request)
    return response

@app.get("/")
async def root():
    return {"ip": client_ip.get()}

  5. Dado o seguinte código, o que será impresso? Explique o resultado.

    import asyncio
import contextvars

var = contextvars.ContextVar('var', default='padrão')

async def altera():
    var.set('alterado')
    await asyncio.sleep(0.1)
    print(f'dentro: {var.get()}')

async def main():
    print(f'antes: {var.get()}')
    task = asyncio.create_task(altera())
    await asyncio.sleep(0.05)
    print(f'depois: {var.get()}')
    await task

asyncio.run(main())

    ✓ Resposta:

    Será impresso:

    antes: padrão
depois: padrão
dentro: alterado

    Isso ocorre porque a tarefa altera() tem seu próprio contexto, herdado do contexto em que foi criada, mas as alterações dentro dela não afetam o contexto externo. Então, var.get() no main continua sendo 'padrão'.