Что такое contextvars в Python и для чего они используются?

Contextvars в Python

Contextvars — это модуль Python (введён в версии 3.7), который предоставляет механизм для управления переменными контекста, специфичными для асинхронных задач и потоков. Это решение для проблемы, когда нужно хранить данные, которые логически принадлежат определённому контексту выполнения, но не могут быть переданы как параметры через весь стек вызовов.

Основные концепции

Проблема, которую решает contextvars

В многопоточных и асинхронных приложениях часто нужно сохранять контекстную информацию (user_id, request_id, трейсинг-данные). Обычные глобальные переменные в этом случае небезопасны:

# ❌ Плохо — не потокобезопасно
global_user_id = None

def process_request(user_id):
    global global_user_id
    global_user_id = user_id
    result = some_operation()
    return result

В многопоточной среде разные потоки перепишут друг другу значение.

Правильное решение с contextvars

import contextvars

# Создаём контекстную переменную
user_id_var = contextvars.ContextVar("user_id", default=None)

def process_request(user_id):
    token = user_id_var.set(user_id)  # Установливаем значение
    try:
        result = some_operation()
        return result
    finally:
        user_id_var.reset(token)  # Восстанавливаем прежнее значение

def some_operation():
    # Получаем значение из контекста, без передачи как параметра
    current_user = user_id_var.get()
    print(f"Processing for user: {current_user}")

Применение в асинхронном коде

Contextvars особенно полезны в asyncio, где каждая задача имеет свой контекст:

import asyncio
import contextvars

request_id = contextvars.ContextVar("request_id")

async def fetch_data(user_id):
    request_id.set(f"req-{user_id}-{id(asyncio.current_task())}")
    await asyncio.sleep(1)
    print(f"Request ID: {request_id.get()}")

async def main():
    # Каждая задача имеет свой контекст
    await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3),
    )

Практическое применение

В веб-фреймворках (FastAPI, Flask)

from fastapi import FastAPI, Request
import contextvars
import logging

app = FastAPI()
request_id = contextvars.ContextVar("request_id", default=None)

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    req_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    token = request_id.set(req_id)
    
    # Логи автоматически могут использовать request_id
    response = await call_next(request)
    request_id.reset(token)
    return response

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Получаем request_id без параметров
    req_id = request_id.get()
    logger.info(f"Fetching user {user_id}, request: {req_id}")
    return {"user_id": user_id}

В логировании

import logging
import contextvars

user_context = contextvars.ContextVar("user", default="anonymous")

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.user = user_context.get()
        return True

logger = logging.getLogger(__name__)
logger.addFilter(ContextFilter())

formatter = logging.Formatter("%(asctime)s - %(user)s - %(message)s")
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)

user_context.set("john_doe")
logger.info("User action performed")  # Выведет: User action performed for john_doe

Копирование контекста

Есть также методы для копирования контекста при создании новых потоков или задач:

import contextvars
import asyncio

var = contextvars.ContextVar("var")

async def main():
    var.set("original")
    
    # Скопируем контекст
    ctx = contextvars.copy_context()
    
    # Запустим функцию в копированном контексте
    result = await asyncio.to_thread(blocking_func)

Ключевые преимущества

• Потокобезопасность: Каждый поток/задача имеет свою копию значения
• Чистота кода: Не нужно передавать контекст через все функции
• Встроенная поддержка asyncio: Автоматически работает с асинхронным кодом
• Trace-ID и Logging: Идеально для распределённого трейсинга
• Type-safe: Используют типизацию через ContextVar[T]