Python のコンテキスト変数（contextvars）と非同期処理

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