contextvars --- 上下文變量?

上下文變量?

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

此類用于聲明一個(gè)新的上下文變量，如:

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

name 參數(shù)用于內(nèi)省和調(diào)試，必需。

調(diào)用 ContextVar.get() 時(shí)，如果上下文中沒有找到此變量的值，則返回可選的僅命名參數(shù) default 。

重要： 上下文變量應(yīng)該在頂級(jí)模塊中創(chuàng)建，且永遠(yuǎn)不要在閉包中創(chuàng)建。 Context 對(duì)象擁有對(duì)上下文變量的強(qiáng)引用，這可以讓上下文變量被垃圾收集器正確回收。

name?

上下文變量的名稱，只讀屬性。

3.7.1 新版功能.

get([default])?

返回當(dāng)前上下文中此上下文變量的值。

如果當(dāng)前上下文中此變量沒有值，則此方法會(huì):

• 如果提供了得 話，返回傳入的 default 值；或者

• 返回上下文變量本身的默認(rèn)值， 如果創(chuàng)建此上下文變量時(shí)提供了默認(rèn)值；或者

• 拋出 LookupError 異常。

set(value)?

調(diào)用此方法設(shè)置上下文變量在當(dāng)前上下文中的值。

必選參數(shù) value 是上下文變量的新值。

返回一個(gè) Token  對(duì)象，可通過 ContextVar.reset() 方法將上下文變量還原為之前某個(gè)狀態(tài)。

reset(token)?

將上下文變量重置為調(diào)用 ContextVar.set() 之前、創(chuàng)建 token 時(shí)候的狀態(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 對(duì)象。此對(duì)象可以傳遞給 ContextVar.reset() 方法用于將上下文變量還原為調(diào)用 set 前的狀態(tài)。

var?

只讀屬性。指向創(chuàng)建此 token 的 ContextVar 對(duì)象。

old_value?

一個(gè)只讀屬性。 會(huì)被設(shè)為在創(chuàng)建此令牌的 ContextVar.set() 方法調(diào)用之前該變量所具有的值。 如果調(diào)用之前變量沒有設(shè)置值，則它指向 Token.MISSING 。

MISSING?

Token.old_value 會(huì)用到的一個(gè)標(biāo)記對(duì)象。

手動(dòng)上下文管理?

contextvars.copy_context()?

返回當(dāng)前上下文中 Context 對(duì)象的拷貝。

以下代碼片段會(huì)獲取當(dāng)前上下文的拷貝并打印設(shè)置到其中的所有變量及其值:

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

此函數(shù)復(fù)雜度為 O(1) ，也就是說對(duì)于只包含幾個(gè)上下文變量和很多上下文變量的情況，他們是一樣快的。

class contextvars.Context?

ContextVars 中所有值的映射。

Context() 創(chuàng)建一個(gè)不包含任何值的空上下文。如果要獲取當(dāng)前上下文的拷貝，使用 copy_context() 函數(shù)。

Context 實(shí)現(xiàn)了 collections.abc.Mapping 接口。

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

按照 run 方法中的參數(shù)在上下文對(duì)象中執(zhí)行 callable(*args, **kwargs) 代碼。返回執(zhí)行結(jié)果，如果發(fā)生異常，則將異常透?jìng)鞒鰜怼?/p>

callable 對(duì)上下文變量所做的任何修改都會(huì)保留在上下文對(duì)象中:

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)在多個(gè)系統(tǒng)線程或者遞歸調(diào)用同一個(gè)上下文對(duì)象的此方法，拋出 RuntimeError 異常。

copy()?

返回此上下文對(duì)象的淺拷貝。

var in context

如果*context* 中含有名稱為 var 的變量，返回 True ， 否則返回 False 。

context[var]

返回名稱為 var 的 ContextVar 變量。如果上下文對(duì)象中不包含這個(gè)變量，則拋出 KeyError 異常。

get(var[, default])?

如果 var 在上下文對(duì)象中具有值則返回 var 的值。 在其他情況下返回 default。 如果未給出 default 則返回 None。

iter(context)

返回一個(gè)存儲(chǔ)在上下文對(duì)象中的變量的迭代器。

len(proxy)

返回上下文對(duì)象中所設(shè)的變量的數(shù)量。

keys()?

返回上下文對(duì)象中的所有變量的列表。

values()?

返回上下文對(duì)象中所有變量值的列表。

items()?

返回包含上下文對(duì)象中所有變量及其值的 2 元組的列表。

asyncio 支持?

上下文變量在 asyncio 中有原生的支持并且無需任何額外配置即可被使用。 例如，以下是一個(gè)簡(jiǎn)單的回顯服務(wù)器，它使用上下文變量來讓遠(yuǎn)程客戶端的地址在處理該客戶端的 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}\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(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