Python contextvars 實戰：Async Context、結構化 Logging 與隔離測試

Python 學習 - 本文屬於一個選集。

§ 132: 本文

一. 前言：參數沒有傳，狀態卻不能亂跑
#

一個 API 請求進來後，很多層程式都會想知道同一批資訊：

request ID；
使用者或租戶 ID；
locale；
tracing metadata；
本次工作的 deadline。最明白的作法，是把 request_id 當參數一路傳下去。但當它穿過十幾層函式、logging adapter、callback 和非同步 task，函式簽名很快就只剩「搬運狀態」的雜訊。全域變數也不行。兩個請求同時執行時，後寫入的值會覆蓋前一個，log 馬上認錯人。 threading.local() 在傳統多執行緒程式很好用，但多個 asyncio.Task 可以輪流跑在同一條 thread；只用 thread-local 無法把它們分開。這正是標準庫 contextvars 的工作：讓狀態跟著目前的執行 context，而不是跟著整個 process 或單一 thread。這篇不重講 async/await 或 logging 基礎。拍拍君會專注回答五個實務問題：

ContextVar 的值究竟放在哪裡？
為什麼 set() 後一定要保留 Token？
asyncio task 何時複製 context？
跨 thread 時哪些 API 會傳遞 context？
如何證明兩個請求沒有互相污染？

二. 安裝與心智模型
#

contextvars 從 Python 3.7 起就是標準庫，不必安裝套件：

python --version
python -c "import contextvars; print(contextvars.__file__)"

如果要執行本文測試，再準備 pytest：

python -m venv .venv
source .venv/bin/activate
python -m pip install -U pytest pytest-asyncio

先比較三種儲存位置：

機制	隔離單位	適合用途	非同步 task 安全？
module global	process	唯讀設定、常數	❌ 可變狀態會互撞
`threading.local()`	OS thread	傳統 thread worker 狀態	❌ 同 thread 的 tasks 仍共用
`ContextVar`	current context	request/task scoped metadata	✅ `asyncio` 原生支援
關鍵句是：`ContextVar` 物件不是值的容器，而是查詢目前 context 的 key。
同一個 `request_id_var`，在不同 context 中可以查到不同值。
因此 `ContextVar` 應放在 module 頂層，所有需要它的程式都 import 同一個 key：

# request_context.py
from contextvars import ContextVar
request_id_var: ContextVar[str] = ContextVar(
    "request_id",
    default="-",
)

不要在每次請求、closure 或函式內建立新的 ContextVar。 Context 會持有變數的強引用；不斷建立短命 key，既失去共享查詢的意義，也可能讓它們無法如預期回收。

三. 第一個 ContextVar：get、set 與 default
#

先看最小範例：

from contextvars import ContextVar
theme_var: ContextVar[str] = ContextVar("theme", default="light")
print(theme_var.get())
theme_var.set("dark")
print(theme_var.get())

輸出是：

light
dark

取得值時，優先順序如下：

目前 context 已經設定的值；
get(default) 當次呼叫提供的預設值；
建立 ContextVar 時提供的 default；
都沒有時，拋出 LookupError。

from contextvars import ContextVar
required_user: ContextVar[str] = ContextVar("required_user")
print(required_user.get("anonymous"))
try:
    required_user.get()
except LookupError:
    print("目前 context 還沒有 user")

對 request ID 而言，"-" 很適合當 logging fallback。對權限或租戶資訊，拍拍君更偏好不給 default，讓漏設狀態立刻失敗，而不是悄悄使用錯誤身分。

四. Token 與 reset：把狀態限制在正確範圍
#

set() 不只改值，還會回傳一個 Token。這個 token 記得「改之前的狀態」，可以交給 reset() 復原：

from contextvars import ContextVar
mode_var: ContextVar[str] = ContextVar("mode", default="normal")
outer_token = mode_var.set("request")
print(mode_var.get())
inner_token = mode_var.set("admin-check")
print(mode_var.get())
mode_var.reset(inner_token)
print(mode_var.get())
mode_var.reset(outer_token)
print(mode_var.get())

輸出依序是 request、admin-check、request、normal。正式程式應該把 reset 放在 finally，避免 exception 或 cancellation 留下髒狀態：

from contextvars import ContextVar
from collections.abc import Iterator
from contextlib import contextmanager
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

@contextmanager
def request_scope(request_id: str) -> Iterator[None]:
    token = request_id_var.set(request_id)
    try:
        yield
    finally:
        request_id_var.reset(token)

使用時就像一般 context manager：

print(request_id_var.get())
with request_scope("req-7f31"):
    print(request_id_var.get())
print(request_id_var.get())

這裡最重要的不是少打一個參數，而是 scope 清楚：進入時綁定，離開時無條件復原。 Python 3.14 起，Token 本身也支援 context manager protocol；若專案仍支援較舊 Python，try/finally 或上面的 wrapper 最直觀。

五. asyncio 如何傳遞 context？
#

asyncio 會在建立 Task 時複製目前 context。子 task 之後改自己的值，不會回寫父 task，也不會污染 sibling task。

import asyncio
from contextvars import ContextVar
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

async def worker(label: str) -> tuple[str, str, str]:
    inherited = request_id_var.get()
    token = request_id_var.set(f"{inherited}:{label}")
    try:
        await asyncio.sleep(0)
        local = request_id_var.get()
    finally:
        request_id_var.reset(token)
    return label, inherited, local

async def main() -> None:
    parent_token = request_id_var.set("req-parent")
    try:
        first = asyncio.create_task(worker("A"))
        second = asyncio.create_task(worker("B"))
        print(await asyncio.gather(first, second))
        print("parent:", request_id_var.get())
    finally:
        request_id_var.reset(parent_token)

asyncio.run(main())

結果會包含：

[('A', 'req-parent', 'req-parent:A'), ('B', 'req-parent', 'req-parent:B')]
parent: req-parent

這裡有三個語意要抓住：

task 建立時取得父 context 的快照；
task 內的 await 不會讓 context 跑到別的 task；
context copy 是隔離 mapping，不代表裡面的可變物件會 deep-copy。最後一點很容易踩雷。不要把共用 list 或 dict 放進 ContextVar 後，在不同 task 原地修改它。優先存字串、整數、不可變 dataclass，或在更新時建立新物件。

六. Snapshot 時機：先 create_task，後 set 已經來不及
#

下面的順序看似接近，結果卻不同：

import asyncio
from contextvars import ContextVar
label_var: ContextVar[str] = ContextVar("label", default="unset")

async def read_later() -> str:
    await asyncio.sleep(0)
    return label_var.get()

async def main() -> None:
    token = label_var.set("before-create")
    try:
        task = asyncio.create_task(read_later())
        label_var.set("after-create")
        print(await task)
    finally:
        label_var.reset(token)

asyncio.run(main())

它印出 before-create，因為 context 在 create_task() 當下就複製了，不是 coroutine 第一次真正執行時才複製。需要刻意用乾淨 context 啟動背景工作時，可以明確傳入 context=：

import asyncio
from contextvars import Context

async def launch_clean() -> str:
    task = asyncio.create_task(
        read_later(),
        name="clean-worker",
        context=Context(),
    )
    return await task

context= 是 Python 3.11 加入的參數。它適合不應繼承 request 身分的長壽命工作；不要只是為了「看起來乾淨」就抹掉 tracing 或 deadline，先定義清楚 propagation policy。

七. 實戰：把 request ID 注入標準 logging
#

logging.Filter 可以在每筆 LogRecord 輸出前加入 context 欄位。這讓深層函式照常呼叫 logger.info()，不必每層都傳 extra={...}：

import logging
from contextvars import ContextVar
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

class RequestContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True

def configure_logging() -> logging.Logger:
    handler = logging.StreamHandler()
    handler.addFilter(RequestContextFilter())
    handler.setFormatter(
        logging.Formatter(
            "%(asctime)s %(levelname)s request_id=%(request_id)s %(message)s"
        )
    )
    logger = logging.getLogger("拍拍君.service")
    logger.handlers.clear()
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    return logger

接著把 scope 放在真正的 request boundary：

import asyncio
logger = configure_logging()

async def load_profile() -> None:
    await asyncio.sleep(0.01)
    logger.info("profile loaded")

async def handle_request(request_id: str) -> None:
    with request_scope(request_id):
        logger.info("request started")
        await load_profile()
        logger.info("request finished")

async def main() -> None:
    await asyncio.gather(
        handle_request("req-alpha"),
        handle_request("req-beta"),
    )

asyncio.run(main())

兩個請求可以交錯執行，但每行 log 都會取得所屬 task 的 request ID。這篇使用標準庫 logging.Filter 展示機制；若你需要完整 handler、rotation 與 dictConfig，可以搭配 Python Logging 教學閱讀。如果系統已有 OpenTelemetry，request ID 不應取代 trace/span correlation；它們可以是不同欄位，各自服務客服查詢與分散式追蹤。

八. 跨 thread 邊界：to_thread 與 executor 不一樣
#

ContextVar 解決 context-local state，但不代表所有 thread API 都會自動複製 context。 asyncio.to_thread() 會把目前 context 傳到 worker thread：

import asyncio

def blocking_read() -> str:
    return request_id_var.get()

async def main() -> None:
    with request_scope("req-thread"):
        inherited = await asyncio.to_thread(blocking_read)
        print(inherited)

asyncio.run(main())

輸出是 req-thread。但直接呼叫 ThreadPoolExecutor.submit() 時，不要假設會繼承提交者的 context。需要傳遞時，先 copy_context()，再讓 worker 在該 context 內執行：

from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context

def run_in_pool() -> str:
    context = copy_context()
    with ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(context.run, request_id_var.get)
        return future.result()

with request_scope("req-pool"):
    print(run_in_pool())

copy_context() 是 O(1)，但 copy 仍然是 shallow mapping。而且同一個 Context 不能同時在多條 thread 裡重複 enter。每個並行 job 應取得自己的 copy，不要把一份 context 當成共享容器四處重用。

九. 隔離測試：同時驗證繼承、分流與清理
#

測試不能只看「值拿得到」，還要證明：

每個 task 繼承正確 request ID；
sibling task 的修改彼此看不到；
scope 結束後回到原值；
exception 發生時仍會 reset。

# test_request_context.py
import asyncio
import pytest
from request_context import request_id_var, request_scope

@pytest.mark.asyncio
async def test_sibling_tasks_are_isolated() -> None:
    async def child(suffix: str) -> tuple[str, str]:
        inherited = request_id_var.get()
        token = request_id_var.set(f"{inherited}:{suffix}")
        try:
            await asyncio.sleep(0)
            return inherited, request_id_var.get()
        finally:
            request_id_var.reset(token)
    with request_scope("req-test"):
        results = await asyncio.gather(child("A"), child("B"))
    assert results == [
        ("req-test", "req-test:A"),
        ("req-test", "req-test:B"),
    ]
    assert request_id_var.get() == "-"

def test_scope_resets_after_exception() -> None:
    with pytest.raises(RuntimeError):
        with request_scope("req-fail"):
            raise RuntimeError("boom")
    assert request_id_var.get() == "-"

執行：

pytest -q

若測試會直接呼叫 set()，也要在 fixture teardown 或 finally reset token。不要依賴「pytest 下一個 test 應該會換環境」；同一條執行 context 裡，沒有 reset 的值就可能留下來。

十. 常見錯誤與 code review 清單
#

1. 把 ContextVar 當萬用 dependency injection
#

資料庫連線、service object、業務輸入仍應明確傳參數。 ContextVar 最適合跨層但與 request scope 綁定的 metadata，不是用來藏所有相依物件。

2. set 了卻不 reset
#

在同步 handler、測試或重用 task 的 worker loop 中，後續工作可能讀到舊值。規則很簡單：拿到 token，就在 finally reset，或包成 scope context manager。

3. 在 ContextVar 內原地修改 mutable value
#

task context 的 mapping 會複製，但 value 不會自動 deep-copy。把 list append 進去，可能仍是在改同一個 list。優先使用 immutable value，更新時整個替換。

4. 猜測 background task 的繼承時機
#

context 在 task 建立時複製。先建立 task 再改值，task 不會看到後來的修改。在 create_task() 附近完成必要綁定，或明確傳入 context=。

5. 假設所有 thread API 都自動 propagation
#

asyncio.to_thread() 會傳遞目前 context；raw executor 或其他 queue/worker 邊界要查 API 契約。需要時使用 copy_context().run(...)，並為每個並行工作建立獨立 copy。拍拍君的 review checklist：

ContextVar 定義在 module 頂層；
名稱可以在除錯時辨識；
安全 fallback 才給 default；
每次 set() 都有對應 reset()；
request scope 放在入口邊界；
background task 的繼承政策有文件與測試；
thread/process/message queue 邊界明確處理 propagation；
context value 避免可變共享物件；
log 不放密碼、token 或敏感個資。

十一. ContextVar、函式參數與 tracing 怎麼選？
#

需求	建議
業務函式真正需要的資料	明確函式參數
單一 request 的 correlation metadata	`ContextVar`
傳統同步 thread worker 的私有狀態	視情況用 `threading.local()`
跨服務 trace/span 傳遞	OpenTelemetry 等 tracing 系統
跨 process 或 message broker	明確序列化進 header/message
`ContextVar` 不會自動穿越 process、HTTP 或 message queue。
它只管理 Python 執行 context 內的狀態；離開這個邊界後，仍要把必要資訊放進協定，再由接收端驗證並綁到新的 context。
想理解一般 task、timeout 與 cancellation，可以先看 Python asyncio 入門；需要結構化並行與跨框架工具，則參考 Python AnyIO 實戰。

結語
#

contextvars 的 API 很少，真正困難的是 propagation policy。記住四件事就夠了：

ContextVar 是目前 context 的查詢 key，應定義在 module 頂層；
set() 回傳的 token 要在離開 scope 時 reset；
asyncio task 在建立時複製 context，之後彼此隔離；
跨 thread、process 與網路邊界時，必須確認或明確實作傳遞。當 scope、邊界與測試都寫清楚，request ID 和 logging metadata 就能安靜地跟著正確工作走，不必污染每一層函式簽名，也不會在兩個同時執行的請求之間串台。