В мире потоков всё было просто: 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доступен через ContextVarStarlette/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 без магии: практический разбор для питонистов». Записаться
