contextvars --- 情境變數


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

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

其他詳細資訊,請參閱PEP 567

在 3.7 版被加入.

情境變數

classcontextvars.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。
classcontextvars.Token

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

The token supportscontext manager protocolto restore the corresponding context variable value at the exit fromwith block:

var=ContextVar('var',default='default value')withvar.set('new value'):assertvar.get()=='new value'assertvar.get()=='default value'

在 3.14 版被加入:新增對用作情境管理器的支援。

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) 的複雜度,也就是說,對於只有少許情境變數的情境和有大量情境變數的情境,速度都一樣快。

classcontextvars.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),則傳播例外。

例如:

importcontextvarsvar=contextvars.ContextVar('var')var.set('spam')print(var.get())# 'spam'ctx=contextvars.copy_context()defmain():# '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)。

varincontext

如果情境裡面有var 的值,則回傳True,否則回傳False

context[var]

回傳varContextVar 變數的值。如果該變數並沒有在情境物件中設定,則會引發KeyError 錯誤。

get(var[,default])

如果var 的值在情境物件中,則回傳var 的值。否則回傳default。如果沒有default 值,則回傳None

iter(context)

回傳儲存於情境物件中變數的疊代器。

len(proxy)

回傳情境物件中的變數個數。

keys()

回傳情境物件中所有變數的串列。

values()

回傳情境物件中所有變數的值的串列。

items()

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

對 asyncio 的支援

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

importasyncioimportcontextvarsclient_addr_var=contextvars.ContextVar('client_addr')defrender_goodbye():# 即使不把目前處理中的用戶端(client)傳入此函式# 仍可取得其位址client_addr=client_addr_var.get()returnf'Good bye, client @{client_addr}\r\n'.encode()asyncdefhandle_request(reader,writer):addr=writer.transport.get_extra_info('socket').getpeername()client_addr_var.set(addr)# 在任何我們呼叫的程式碼中,都可以# 呼叫 'client_addr_var.get()' 來取得用戶端的位址whileTrue:line=awaitreader.readline()print(line)ifnotline.strip():breakwriter.write(b'HTTP/1.1 200 OK\r\n')# status linewriter.write(b'\r\n')# headerswriter.write(render_goodbye())# bodywriter.close()asyncdefmain():srv=awaitasyncio.start_server(handle_request,'127.0.0.1',8081)asyncwithsrv:awaitsrv.serve_forever()asyncio.run(main())# 你可以使用 telnet 或 curl 測試:#     telnet 127.0.0.1 8081#     curl 127.0.0.1:8081