`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* 呼叫之前的值。

token 支援上下文管理器協議，以便在退出 with 程式碼塊時恢復相應的上下文變數值。

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

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

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

在 3.14 版本加入: 添加了作為上下文管理器的用法支援。

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` — 上下文變數¶