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 はロギング、認証情報、データベース接続など、リクエストスコープの状態管理に最適です。
