В мире потоков всё было просто: threading.local() даёт каждому потоку свои данные. Request ID, текущий пользователь, database connection — положил в thread-local, достал когда нужно. FastAPI, Flask, Django — все так делали.

Потом пришёл asyncio, и эта модель сломалась. В одном потоке выполняются тысячи корутин, и thread-local у них общий. Положил request ID в одной корутине — прочитал чужой в другой. contextvars, появившийся в Python 3.7, решает эту проблему, но механика его работы не очевидна.

Разберём, почему thread-local не работает в async, как устроены contextvars, и какие паттерны использовать.

Проблема: thread-local в asyncio

Thread-local variables — данные, уникальные для каждого потока:

import threading

local = threading.local()

def handle_request(request_id):
    local.request_id = request_id
    process()
    
def process():
    print(f"Processing {local.request_id}")

В синхронном сервере каждый request обрабатывается в своём потоке. Thread-local хранит контекст request, все функции в call stack имеют к нему доступ без передачи аргументов.

Теперь asyncio:

import asyncio
import threading

local = threading.local()

async def handle_request(request_id):
    local.request_id = request_id
    await asyncio.sleep(0.1)  # Симуляция I/O
    print(f"Request {request_id}, got {local.request_id}")

async def main():
    await asyncio.gather(
        handle_request("A"),
        handle_request("B"),
        handle_request("C"),
    )

asyncio.run(main())

Вывод:

Request A, got C
Request B, got C
Request C, got C

Все три корутины выполняются в одном потоке. Когда корутина A засыпает на await, управление переходит к B, потом к C. Каждая перезаписывает local.request_id. Когда A просыпается, там уже значение от C.

Thread-local привязан к потоку, а не к корутине. В async это бесполезно.

ContextVar: thread-local для корутин

contextvars.ContextVar решает проблему:

import asyncio
from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar('request_id', default='unknown')

async def handle_request(rid):
    request_id.set(rid)
    await asyncio.sleep(0.1)
    print(f"Request {rid}, got {request_id.get()}")

async def main():
    await asyncio.gather(
        handle_request("A"),
        handle_request("B"),
        handle_request("C"),
    )

asyncio.run(main())

Вывод:

Request A, got A
Request B, got B
Request C, got C

Каждая корутина видит своё значение, несмотря на общий поток. Как это вообще работает?

Context и copy-on-write

Основная структура — Context. Это immutable mapping из ContextVar в значения. У каждого таска asyncio свой Context.

При создании Task копируется текущий Context:

async def outer():
    request_id.set("outer")
    
    # Создаём task — он получает КОПИЮ текущего контекста
    task = asyncio.create_task(inner())
    await task

async def inner():
    # Видим значение из момента создания task
    print(request_id.get())  # "outer"
    
    # Изменяем — это изменение локально для inner
    request_id.set("inner")

async def main():
    await outer()
    # После завершения outer контекст main не изменился

Но копирование не означает дублирование всех данных. Используется copy-on-write: Context ссылается на родительский, пока не происходит изменение. При set() создаётся новый узел только для изменённой переменной.

Это похоже на persistent data structures: изменение создаёт новую версию, разделяющую неизменённые части со старой.

Структура Context

Посмотрим на Context напрямую:

from contextvars import ContextVar, copy_context

var1: ContextVar[int] = ContextVar('var1')
var2: ContextVar[str] = ContextVar('var2')

var1.set(42)
var2.set("hello")

# Получаем текущий контекст
ctx = copy_context()

print(list(ctx.items()))
# [(var1, 42), (var2, 'hello')]

# Context — это mapping
print(ctx[var1])  # 42
print(var1 in ctx)  # True

copy_context() возвращает снапшот текущего контекста. Это не живая ссылка, изменения после копирования не видны.

Context можно использовать для запуска функций в изолированном окружении:

def worker():
    print(f"var1 = {var1.get()}")
    var1.set(100)
    print(f"var1 after set = {var1.get()}")

ctx = copy_context()

# Запускаем worker в контексте ctx
ctx.run(worker)
# var1 = 42
# var1 after set = 100

# Изменения остались внутри ctx
print(var1.get())  # 42 — оригинальный контекст не изменился
print(ctx[var1])   # 100 — но ctx изменился

ctx.run(func) — основной метод. Он временно устанавливает ctx как текущий контекст, запускает функцию, потом восстанавливает предыдущий. Все изменения ContextVar внутри func применяются к ctx.

Как asyncio использует Context

При создании Task asyncio делает примерно следующее:

class Task:
    def __init__(self, coro):
        self._coro = coro
        self._context = copy_context()  # Снапшот при создании
    
    def _step(self):
        # Выполняем шаг корутины в её контексте
        self._context.run(self._coro.send, None)

Каждый Task имеет свой Context. При переключении между задачами Context переключается автоматически.

Именно поэтому asyncio.create_task() захватывает контекст:

async def example():
    request_id.set("parent")
    
    # Task создаётся с копией текущего контекста
    task = asyncio.create_task(child())
    
    # Изменения в parent после создания task...
    request_id.set("parent_modified")
    
    await task

async def child():
    # ...не видны в child — он работает со снапшотом
    print(request_id.get())  # "parent", не "parent_modified"

Task получает контекст на момент создания, а не на момент первого await.

Token: откат изменений

set() возвращает Token, позволяющий откатить изменение:

token = request_id.set("temporary")
try:
    do_something()
finally:
    request_id.reset(token)  # Возвращаем предыдущее значение

Это полезно для временного изменения контекста, например, в middleware:

async def logging_middleware(request, handler):
    token = request_id.set(request.headers.get("X-Request-ID"))
    try:
        return await handler(request)
    finally:
        request_id.reset(token)

Без Token пришлось бы запоминать предыдущее значение вручную, что ненадёжно при исключениях.

Логирование с request ID

Такой вот базовый пример — добавление request ID во все логи:

import logging
from contextvars import ContextVar
from typing import Optional

request_id: ContextVar[Optional[str]] = ContextVar('request_id', default=None)

class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id.get() or 'no-request'
        return True

# Настройка логгера
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    '%(asctime)s [%(request_id)s] %(message)s'
))
handler.addFilter(RequestIdFilter())

logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Использование
async def handle_request(rid: str):
    request_id.set(rid)
    logger.info("Starting request")
    await do_work()
    logger.info("Finished request")

async def do_work():
    # Не передаём request_id явно — он в контексте
    logger.info("Doing work")
    await asyncio.sleep(0.1)

Вывод:

2024-01-15 10:00:00 [req-123] Starting request
2024-01-15 10:00:00 [req-123] Doing work
2024-01-15 10:00:00 [req-123] Finished request

Request ID пробрасывается через весь call stack без явной передачи.

database connection per request

Ещё один частый паттерн — connection pooling с привязкой к request:

from contextvars import ContextVar
from typing import Optional
import asyncpg

db_connection: ContextVar[Optional[asyncpg.Connection]] = ContextVar(
    'db_connection', default=None
)

class Database:
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def get_connection(self) -> asyncpg.Connection:
        conn = db_connection.get()
        if conn is None:
            raise RuntimeError("No database connection in context")
        return conn
    
    async def transaction(self):
        """Context manager для транзакции"""
        async with self.pool.acquire() as conn:
            token = db_connection.set(conn)
            try:
                async with conn.transaction():
                    yield conn
            finally:
                db_connection.reset(token)

# Использование
db = Database(pool)

async def handle_request():
    async with db.transaction():
        await create_user()
        await send_notification()
        # Если exception — откат транзакции

async def create_user():
    conn = await db.get_connection()  # Получаем из контекста
    await conn.execute("INSERT INTO users ...")

async def send_notification():
    conn = await db.get_connection()  # Та же connection
    await conn.execute("INSERT INTO notifications ...")

Все функции внутри transaction() используют одну connection без явной передачи. Транзакция атомарна.

Интеграция с thread pools

run_in_executor автоматически копирует контекст:

async def example():
    request_id.set("async-context")
    
    loop = asyncio.get_running_loop()
    
    # Контекст копируется в thread
    result = await loop.run_in_executor(
        None, 
        blocking_function
    )

def blocking_function():
    # Видим значение из async-контекста
    print(request_id.get())  # "async-context"

Это работает начиная с Python 3.7. Раньше приходилось передавать контекст явно.

Для ThreadPoolExecutor без asyncio:

from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context

def run_with_context(func, *args):
    ctx = copy_context()
    return ctx.run(func, *args)

executor = ThreadPoolExecutor()

request_id.set("main")

# Запускаем с копированием контекста
future = executor.submit(run_with_context, worker)

Проблемы(куда без них)

Изменения не видны в родительском контексте:

async def parent():
    var.set("parent")
    await asyncio.create_task(child())
    print(var.get())  # Всё ещё "parent"!

async def child():
    var.set("child")  # Изменяем копию контекста

Task работает с копией. Изменения в child не влияют на parent. Если нужно передать данные обратно — используйте return или явные структуры (Queue, Event).

Callback в loop.call_soon:

async def example():
    request_id.set("example")
    
    loop = asyncio.get_running_loop()
    
    # call_soon НЕ копирует контекст автоматически (до Python 3.11)
    loop.call_soon(callback)  # callback увидит другой контекст
    
    # Явная передача контекста
    ctx = copy_context()
    loop.call_soon(callback, context=ctx)

В Python 3.11+ появился параметр context= для call_soon, call_later, etc.

ContextVar в классах:

class Service:
    # Не делайте так — один экземпляр ContextVar на все объекты
    request_id = ContextVar('request_id')

# Делайте так — ContextVar на уровне модуля
request_id: ContextVar[str] = ContextVar('request_id')

class Service:
    def get_request_id(self):
        return request_id.get()

ContextVar должен быть синглтоном. Создание ContextVar в init каждого объекта — ошибка.

Тяжёлые объекты в контексте:

Context копируется при создании Task. Если в ContextVar лежит большой объект — копируется ссылка, не объект. Но если много ContextVar с маленькими объектами — overhead складывается.

Храните в ContextVar идентификаторы или ссылки, не большие структуры данных.

Сравнение с альтернативами

threading.local:

• Работает только в sync-коде

• Нет изоляции между корутинами

• Проще в использовании, если не нужен async

Явная передача параметров:

• Самый явный и понятный способ

• Засоряет сигнатуры функций

• Нужно протаскивать через весь call stack

Глобальные переменные:

• Не thread-safe, не async-safe

• Подходят только для read-only конфигурации

contextvars:

• Работает в sync и async

• Автоматическая изоляция в asyncio

• Требует понимания механики копирования

Библиотеки, использующие contextvars

• aiohttp — aiohttp.web.Request доступен через ContextVar

• Starlette/FastAPI — request state

• structlog — контекстные переменные для логирования

• SQLAlchemy — async session management

• OpenTelemetry — trace context propagation

Если пишете async-библиотеку с глобальным состоянием — используйте contextvars.

contextvars решает фундаментальную проблему async Python: как хранить данные, привязанные к логическому потоку выполнения, а не к системному потоку.

Главное — понимать, что изменения в дочернем контексте не видны в родительском.

Если хотите не просто «знать про asyncio», а уверенно проектировать продакшен-код на Python, обратите внимание на курс Python Developer. Professional. На нем разбирают практики и инструменты разработки: асинхронщину, паттерны, метапрограммирование, производительность и безопасность — на задачах, похожих на реальные сервисы. Готовы к серьезному обучению? Пройдите вступительный тест.

А чтобы узнать больше о формате обучения и задать вопросы экспертам, приходите на бесплатные демо-уроки:

• 5 февраля в 20:00. «Python и Web Scraping: Извлечение данных из интернета для анализа и автоматизации». Записаться

• 10 февраля в 20:00. «Делаем по красоте: паттерны проектирования в Python-приложениях». Записаться

• 19 февраля в 20:00. «Kafka без магии: практический разбор для питонистов». Записаться