`contextvars` — 上下文變數¶

此模組提供用於管理、儲存和訪問上下文區域性狀態的 API。ContextVar 類用於宣告和使用上下文變數。copy_context() 函式和 Context 類應用於非同步框架中管理當前上下文。

具有狀態的上下文管理器應使用上下文變數而不是 threading.local()，以防止其狀態在併發程式碼中使用時意外地洩露到其他程式碼中。

另請參閱 PEP 567 以瞭解更多詳細資訊。

3.7 版本中新增。

上下文變數¶

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

此類用於宣告新的上下文變數，例如：

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

必需的 *name* 引數用於自省和除錯目的。

可選的僅限關鍵字的 *default* 引數在當前上下文中未找到變數的值時，由 ContextVar.get() 返回。

重要提示：上下文變數應在頂層模組級別建立，而永遠不要在閉包中建立。Context 物件持有對上下文變數的強引用，這會阻止上下文變數被正確地垃圾回收。

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

在 3.7.1 版本中新增。

get([default])¶

返回當前上下文中上下文變數的值。

如果當前上下文中沒有該變數的值，則此方法將

返回方法的 *default* 引數的值（如果提供）；或者
返回上下文變數的預設值（如果建立時指定了預設值）；或者
引發 LookupError。

set(value)¶

呼叫此方法以設定當前上下文中上下文變數的新值。

必需的 *value* 引數是上下文變數的新值。

返回一個 Token 物件，該物件可用於透過 ContextVar.reset() 方法將變數還原為其先前的值。

reset(token)¶

將上下文變數重置為其在建立 *token* 的 ContextVar.set() 使用之前的值。

例如：

var = ContextVar('var')

token = var.set('new value')
# code that uses 'var'; var.get() returns 'new value'.
var.reset(token)

# After the reset call the var has no value again, so
# var.get() would raise a LookupError.

class contextvars.Token¶

Token 物件由 ContextVar.set() 方法返回。可以將它們傳遞給 ContextVar.reset() 方法，以將變數的值恢復到相應 *set* 之前的值。

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

old_value¶: 只讀屬性。設定為在建立 token 的 ContextVar.set() 方法呼叫之前，變數具有的值。如果該變數在呼叫之前未設定，則它指向 Token.MISSING。

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

手動上下文管理¶

contextvars.copy_context()¶

返回當前 Context 物件的副本。

以下程式碼片段獲取當前上下文的副本，並列印其中設定的所有變數及其值

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

此函式的時間複雜度為 O(1)，即對於具有少量上下文變數的上下文和具有大量上下文變數的上下文，它的執行速度一樣快。

class contextvars.Context¶

ContextVars 到其值的對映。

Context() 建立一個不包含任何值的空上下文。要獲取當前上下文的副本，請使用 copy_context() 函式。

每個執行緒都有自己的有效 Context 物件堆疊。當前上下文是當前執行緒堆疊頂部的 Context 物件。堆疊中的所有 Context 物件都被認為是已進入的。

進入上下文，可以透過呼叫其 run() 方法來完成，它透過將上下文推到當前執行緒的上下文堆疊頂部，使其成為當前上下文。

退出當前上下文，可以透過從傳遞給 run() 方法的回撥函式中返回來實現。這會將當前上下文恢復到進入上下文之前的狀態，方法是將上下文從上下文堆疊的頂部彈出。

由於每個執行緒都有自己的上下文堆疊，因此當在不同的執行緒中分配值時，ContextVar 物件的行為類似於 threading.local()。

嘗試進入一個已經進入的上下文（包括在其他執行緒中進入的上下文）會引發 RuntimeError 異常。

退出上下文後，可以稍後重新進入（從任何執行緒）。

透過 ContextVar.set() 方法對 ContextVar 值所做的任何更改都會記錄在當前上下文中。 ContextVar.get() 方法返回與當前上下文關聯的值。退出上下文實際上會還原在進入上下文時對上下文變數所做的任何更改（如果需要，可以透過重新進入上下文來恢復值）。

Context 實現了 collections.abc.Mapping 介面。

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

進入 Context，執行 callable(*args, **kwargs)，然後退出 Context。返回 callable 的返回值，如果發生異常則傳播該異常。

示例

import contextvars

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

ctx = contextvars.copy_context()

def main():
    # 'var' was set to 'spam' before
    # calling 'copy_context()' and 'ctx.run(main)', so:
    print(var.get())  # 'spam'
    print(ctx[var])  # 'spam'

    var.set('ham')

    # Now, after setting 'var' to 'ham':
    print(var.get())  # 'ham'
    print(ctx[var])  # 'ham'

# Any changes that the 'main' function makes to 'var'
# will be contained in 'ctx'.
ctx.run(main)

# The 'main()' function was run in the 'ctx' context,
# so changes to 'var' are contained in it:
print(ctx[var])  # 'ham'

# However, outside of 'ctx', 'var' is still set to 'spam':
print(var.get())  # 'spam'

copy()¶: 返回上下文物件的淺複製。

var in context: 如果 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 元組列表。

asyncio 支援¶

上下文變數在 asyncio 中得到原生支援，無需任何額外配置即可使用。例如，這是一個簡單的回顯伺服器，它使用上下文變數使遠端客戶端的地址在處理該客戶端的 Task 中可用。

import asyncio
import contextvars

client_addr_var = contextvars.ContextVar('client_addr')

def render_goodbye():
    # The address of the currently handled client can be accessed
    # without passing it explicitly to this function.

    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)

    # In any code that we call is now possible to get
    # client's address by calling '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())

# To test it you can use telnet or curl:
#     telnet 127.0.0.1 8081
#     curl 127.0.0.1:8081

`contextvars` — 上下文變數¶

上下文變數¶

手動上下文管理¶

asyncio 支援¶

目錄

上一主題

下一主題

本頁

contextvars — 上下文變數¶

上下文變數¶

手動上下文管理¶

asyncio 支援¶

`contextvars` — 上下文變數¶