contextvars モジュールは、非同期処理でタスクごとに独立した変数を持つ仕組みを提供します。スレッドローカル変数の非同期版です。
問題の背景
グローバル変数や threading.local は、非同期タスク間で意図しない共有が起きます。
import asyncio current_user = None async def process_request(user): global current_user current_user = user await asyncio.sleep(0.1) print(f"Processing for {current_user}") # 別タスクの値になる可能性 async def main(): await asyncio.gather( process_request("Alice"), process_request("Bob"), ) asyncio.run(main())
ContextVar で解決
import asyncio from contextvars import ContextVar current_user: ContextVar[str] = ContextVar("current_user") async def process_request(user): current_user.set(user) await asyncio.sleep(0.1) print(f"Processing for {current_user.get()}") async def main(): await asyncio.gather( process_request("Alice"), process_request("Bob"), ) asyncio.run(main()) # 正しく Alice と Bob が出力される
ContextVar は各タスクのコンテキストに値を保存するため、タスク間で干渉しません。
Token によるリセット
set() は Token を返し、これを使って以前の値に戻せます。
from contextvars import ContextVar var: ContextVar[int] = ContextVar("var", default=0) token = var.set(10) print(var.get()) # 10 var.reset(token) print(var.get()) # 0
実用例：リクエストID のトラッキング
from contextvars import ContextVar import asyncio import uuid request_id: ContextVar[str] = ContextVar("request_id") def log(message): rid = request_id.get("unknown") print(f"[{rid}] {message}") async def handle_request(): request_id.set(str(uuid.uuid4())[:8]) log("Start processing") await asyncio.sleep(0.1) log("Done") async def main(): await asyncio.gather( handle_request(), handle_request(), ) asyncio.run(main())
contextvars はロギング、認証情報、データベース接続など、リクエストスコープの状態管理に最適です。
