新聞中心
contextvars —- 上下文變量
本模塊提供了相關(guān)API用于管理、存儲和訪問上下文相關(guān)的狀態(tài)。 ContextVar 類用于聲明 上下文變量 并與其一起使用。函數(shù) copy_context() 和類 Context 用于管理當(dāng)前上下文和異步框架中。

在多并發(fā)環(huán)境中,有狀態(tài)上下文管理器應(yīng)該使用上下文變量,而不是 threading.local() 來防止他們的狀態(tài)意外泄露到其他代碼。
更多信息參見 PEP 567 。
3.7 新版功能.
上下文變量
class contextvars.ContextVar(name[, **, default*])
此類用于聲明一個新的上下文變量,如:
var: ContextVar[int] = ContextVar('var', default=42)
name 參數(shù)用于內(nèi)省和調(diào)試,必需。
調(diào)用 ContextVar.get() 時,如果上下文中沒有找到此變量的值,則返回可選的僅命名參數(shù) default 。
重要: 上下文變量應(yīng)該在頂級模塊中創(chuàng)建,且永遠不要在閉包中創(chuàng)建。 Context 對象擁有對上下文變量的強引用,這可以讓上下文變量被垃圾收集器正確回收。
-
name
上下文變量的名稱,只讀屬性。
3.7.1 新版功能.
-
get([default])
返回當(dāng)前上下文中此上下文變量的值。
如果當(dāng)前上下文中此變量沒有值,則此方法會:
-
如果提供了得 話,返回傳入的 default 值;或者
-
返回上下文變量本身的默認值, 如果創(chuàng)建此上下文變量時提供了默認值;或者
-
拋出 LookupError 異常。
-
-
set(value)
調(diào)用此方法設(shè)置上下文變量在當(dāng)前上下文中的值。
必選參數(shù) value 是上下文變量的新值。
返回一個 Token 對象,可通過 ContextVar.reset() 方法將上下文變量還原為之前某個狀態(tài)。
-
reset(token)
將上下文變量重置為調(diào)用 ContextVar.set() 之前、創(chuàng)建 token 時候的狀態(tài)。
例如:
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
ContextVar.set() 方法返回 Token 對象。此對象可以傳遞給 ContextVar.reset() 方法用于將上下文變量還原為調(diào)用 set 前的狀態(tài)。
-
var
只讀屬性。指向創(chuàng)建此 token 的 ContextVar 對象。
-
old_value
A read-only property. Set to the value the variable had before the ContextVar.set() method call that created the token. It points to Token.MISSING if the variable was not set before the call.
-
MISSING
Token.old_value 會用到的一個標(biāo)記對象。
手動上下文管理
contextvars.copy_context()
返回當(dāng)前上下文中 Context 對象的拷貝。
以下代碼片段會獲取當(dāng)前上下文的拷貝并打印設(shè)置到其中的所有變量及其值:
ctx: Context = copy_context()print(list(ctx.items()))
此函數(shù)復(fù)雜度為 O(1) ,也就是說對于只包含幾個上下文變量和很多上下文變量的情況,他們是一樣快的。
class contextvars.Context
ContextVars 中所有值的映射。
Context() 創(chuàng)建一個不包含任何值的空上下文。如果要獲取當(dāng)前上下文的拷貝,使用 copy_context() 函數(shù)。
Context 實現(xiàn)了 collections.abc.Mapping 接口。
-
run(callable, \args, **kwargs*)
按照 run 方法中的參數(shù)在上下文對象中執(zhí)行
callable(*args, **kwargs)代碼。返回執(zhí)行結(jié)果,如果發(fā)生異常,則將異常透傳出來。callable 對上下文變量所做的任何修改都會保留在上下文對象中:
var = ContextVar('var')var.set('spam')def main():# 'var' was set to 'spam' before# calling 'copy_context()' and 'ctx.run(main)', so:# var.get() == ctx[var] == 'spam'var.set('ham')# Now, after setting 'var' to 'ham':# var.get() == ctx[var] == 'ham'ctx = copy_context()# 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:# ctx[var] == 'ham'# However, outside of 'ctx', 'var' is still set to 'spam':# var.get() == 'spam'
當(dāng)在多個系統(tǒng)線程或者遞歸調(diào)用同一個上下文對象的此方法,拋出 RuntimeError 異常。
-
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)
返回上下文對象中所設(shè)的變量的數(shù)量。
-
keys()
返回上下文對象中的所有變量的列表。
-
values()
返回上下文對象中所有變量值的列表。
-
items()
返回包含上下文對象中所有變量及其值的 2 元組的列表。
asyncio 支持
上下文變量在 asyncio 中有原生的支持并且無需任何額外配置即可被使用。 例如,以下是一個簡單的回顯服務(wù)器,它使用上下文變量來讓遠程客戶端的地址在處理該客戶端的 Task 中可用:
import asyncioimport contextvarsclient_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}\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():breakwriter.write(line)writer.write(render_goodbye())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:# telnet 127.0.0.1 8081
網(wǎng)頁名稱:創(chuàng)新互聯(lián)Python教程:contextvars—-上下文變量
網(wǎng)站地址:http://m.fisionsoft.com.cn/article/coedpgg.html


咨詢
建站咨詢
