`contextvars` --- 情境變數¶

本模組供 API 來管理、儲存及存取單一情境各自的狀態（context-local state）。用 ContextVar 類別宣告和處理情境變數。 copy_context() 函式和 Context 類別應在非同步框架中管理目前的情境。

帶有狀態的 Context Manager 應該使用情境變數，而不是 threading.local()，才能防止它們的狀態在並行（concurrent）程式碼中使用時意外外溢並干擾到其他程式碼。

其他詳細資訊，請參閱 PEP 567。

在 3.7 版被加入.

情境變數¶

class contextvars.ContextVar(name[, *, default])¶

此類別用在宣告新的情境變數，例如：

var: ContextVar[int] = ContextVar('var', default=42)

必要參數 name 用於自我檢查（introspection）和除錯。

當在目前的情境中找不到變數的值時，ContextVar.get() 會回傳可選的僅限關鍵字參數 default。

重要：情境變數應該在最頂端的模組層級建立，絕對不要在閉包（closure）中建立。 Context 物件持有情境變數的強參照，這會阻止情境變數被正確地垃圾回收（garbage collected）。

name¶: 這個變數的名稱。這是一個唯讀屬性。

在 3.7.1 版被加入.

get([default])¶

回傳當前情境的情境變數值。

如果在當前情境中沒有變數的值，此方法將：

回傳方法的 default 引數值（如果有的話）；否則
回傳情境變數的預設值（如果建立情境變數時有指定預設值的話）；否則
會引發一個 LookupError。

set(value)¶

在目前的情境中，呼叫以設定情境變數的新值。

value 屬必要引數，是情境變數的新值。

回傳一個 Token 物件，該物件可透過 ContextVar.reset() 方法，用來將變數還原到之前的值。

reset(token)¶

將情境變數重設為使用 ContextVar.set() 建立 token 前的值。

舉例來說：

var = ContextVar('var')

token = var.set('new value')
# 使用 'var' 的程式碼；var.get() 回傳 'new value'。
var.reset(token)

# 在重設呼叫之後，var 又沒有值了，所以
# var.get() 會引發 LookupError。

class contextvars.Token¶

Token 物件由 ContextVar.set() 方法回傳，可以傳遞給 ContextVar.reset() 方法，用以將變數的值還原為相對應的 set 之前的值。

The token supports context manager protocol to restore the corresponding context variable value at the exit from with block:

var = ContextVar('var', default='default value')

with var.set('new value'):
    assert var.get() == 'new value'

assert var.get() == 'default value'

在 3.14 版被加入: Added support for usage as a context manager.

var¶: 唯讀屬性。指向建立 token 的 ContextVar 物件。

old_value¶: 唯讀屬性。值為變數在呼叫 ContextVar.set() 方法之前的值。如果變數在呼叫前沒有設定，則指向 Token.MISSING。

MISSING¶: 由 Token.old_value 使用的標記物件。

手動情境管理¶

contextvars.copy_context()¶

回傳目前 Context 物件的複本（copy）。

以下程式碼片段會取得目前情境的複本，並顯示在其中設定的所有變數及其值：:

ctx: Context = copy_context()
print(list(ctx.items()))

這個函式具有 O(1) 的複雜度，也就是說，對於只有少許情境變數的情境和有大量情境變數的情境，速度都一樣快。

class contextvars.Context¶

ContextVars 到其值的映射。

Context() 會建立一個沒有值的空情境。要取得目前情境的複本，請使用 copy_context() 函式。

每個執行緒都有自己的 Context 物件中目前主控中的堆疊（stack）。current context 是目前執行緒堆疊頂端的 Context 物件。堆疊中的所有 Context 物件都被視為已進入。

進入一個情境，可以藉由呼叫其 run() 方法來完成，此進入的動作會將一情境推到目前執行緒的情境堆疊的頂端，使該情境成為目前的情境。

如果你傳遞給 run() 方法的回呼函式（callback functions），該函式回傳之後，就會自動退出目前的情境，這會將目前的情境還原到進入情境之前的狀態，方法是將情境從情境堆疊的頂端彈出。

因為每個執行緒都有自己的情境堆疊，當值在不同的執行緒中被指定時， ContextVar 物件的行為與 threading.local() 相似。

嘗試進入已進入的情境，包括在其他執行緒中進入的情境，會引發 RuntimeError。

退出情境後，之後可以重新進入（從任何執行緒）。

任何透過 ContextVar.set() 方法對 ContextVar 值的改變都會記錄在目前的情境中。 ContextVar.get() 方法回傳與當前情境相關的值。退出情境實際造成的效果會像是將其在進入情境時對情境變數所做的任何變一一彈出並還原（如果需要，可以透過重新進入情境來還原值）。

情境（Context）實作了 collections.abc.Mapping 介面。

run(callable, *args, **kwargs)¶

進入 Context，執行 callable(*args, **kwargs)，然後退出 Context。回傳 callable 的回傳值，如果發生例外（exception），則傳播例外。

例如：

import contextvars

var = contextvars.ContextVar('var')
var.set('spam')
print(var.get())  # 'spam'

ctx = contextvars.copy_context()

def main():
    # 'var' 之前被設成 'spam'
    # 呼叫 'copy_context()' 和 'ctx.run(main)'，所以：
    print(var.get())  # 'spam'
    print(ctx[var])  # 'spam'

    var.set('ham')

    # 現在, 在把 'var' 的值設成 'ham' 後:
    print(var.get())  # 'ham'
    print(ctx[var])  # 'ham'

# 'main' 函式對 'var' 所做的任何變更都會
# 包含在 'ctx 裡：.
ctx.run(main)

# 'main()' 函式是在 'ctx' 情境中執行，
# 所以對 'var' 的變更會保存在 'ctx' 中：
print(ctx[var])  # 'ham'

# 但是，在 'ctx' 外, 'var' 的值仍然是 'spam':
print(var.get())  # 'spam'

copy()¶: 回傳情境物件的淺層複本（shallow copy）。

var in context: 如果情境裡面有 var 的值，則回傳 True，否則回傳 False。

context[var]: 回傳 var ContextVar 變數的值。如果該變數並沒有在情境物件中設定，則會引發 KeyError 錯誤。

get(var[, default])¶: 如果 var 的值在情境物件中，則回傳 var 的值。否則回傳 default。如果沒有 default 值，則回傳 None。

iter(context): 回傳儲存於情境物件中變數的疊代器。

len(proxy): 回傳情境物件中的變數個數。

keys()¶: 回傳情境物件中所有變數的串列。

values()¶: 回傳情境物件中所有變數的值的串列。

items()¶: 回傳情境物件中所有變數與其值的 2-元組（2-tuples）的串列。

對 asyncio 的支援¶

asyncio 原生支援情境變數，不需任何額外設定。舉例來說，以下是一個簡單的 echo 伺服器，使用情境變數讓遠端用戶端的位址在處理該用戶端的任務中可用：

import asyncio
import contextvars

client_addr_var = contextvars.ContextVar('client_addr')

def render_goodbye():
    # 即使不把目前處理中的用戶端（client）傳入此函式
    # 仍可取得其位址

    client_addr = client_addr_var.get()
    return f'Good bye, client @ {client_addr}\r\n'.encode()

async def handle_request(reader, writer):
    addr = writer.transport.get_extra_info('socket').getpeername()
    client_addr_var.set(addr)

    # 在任何我們呼叫的程式碼中，都可以
    # 呼叫 'client_addr_var.get()' 來取得用戶端的位址

    while True:
        line = await reader.readline()
        print(line)
        if not line.strip():
            break

    writer.write(b'HTTP/1.1 200 OK\r\n')  # status line
    writer.write(b'\r\n')  # headers
    writer.write(render_goodbye())  # body
    writer.close()

async def main():
    srv = await asyncio.start_server(
        handle_request, '127.0.0.1', 8081)

    async with srv:
        await srv.serve_forever()

asyncio.run(main())

# 你可以使用 telnet 或 curl 測試:
#     telnet 127.0.0.1 8081
#     curl 127.0.0.1:8081

`contextvars` --- 情境變數¶

情境變數¶

手動情境管理¶

對 asyncio 的支援¶

目錄

上個主題

下個主題

此頁面

contextvars --- 情境變數¶

情境變數¶

手動情境管理¶

對 asyncio 的支援¶

`contextvars` --- 情境變數¶