您好,登錄后才能下訂單哦!
這篇文章主要介紹了Python怎么利用contextvars實(shí)現(xiàn)管理上下文變量的相關(guān)知識(shí),內(nèi)容詳細(xì)易懂,操作簡(jiǎn)單快捷,具有一定借鑒價(jià)值,相信大家閱讀完這篇Python怎么利用contextvars實(shí)現(xiàn)管理上下文變量文章都會(huì)有所收獲,下面我們一起來看看吧。
Python 在 3.7 的時(shí)候引入了一個(gè)模塊:contextvars,從名字上很容易看出它指的是上下文變量(Context Variables),所以在介紹 contextvars 之前我們需要先了解一下什么是上下文(Context)。
Context 是一個(gè)包含了相關(guān)信息內(nèi)容的對(duì)象,舉個(gè)例子:"比如一部 13 集的動(dòng)漫,你直接點(diǎn)進(jìn)第八集,看到女主角在男主角面前流淚了"。相信此時(shí)你是不知道為什么女主角會(huì)流淚的,因?yàn)槟銢]有看前面幾集的內(nèi)容,缺失了相關(guān)的上下文信息。
所以 Context 并不是什么神奇的東西,它的作用就是攜帶一些指定的信息。
我們以 fastapi 和 sanic 為例,看看當(dāng)一個(gè)請(qǐng)求過來的時(shí)候,它們是如何解析的。
# fastapi from fastapi import FastAPI, Request import uvicorn app = FastAPI() @app.get("/index") async def index(request: Request): name = request.query_params.get("name") return {"name": name} uvicorn.run("__main__:app", host="127.0.0.1", port=5555) # ------------------------------------------------------- # sanic from sanic import Sanic from sanic.request import Request from sanic import response app = Sanic("sanic") @app.get("/index") async def index(request: Request): name = request.args.get("name") return response.json({"name": name}) app.run(host="127.0.0.1", port=6666)
發(fā)請(qǐng)求測(cè)試一下,看看結(jié)果是否正確。
可以看到請(qǐng)求都是成功的,并且對(duì)于 fastapi 和 sanic 而言,其 request 和 視圖函數(shù)是綁定在一起的。也就是在請(qǐng)求到來的時(shí)候,會(huì)被封裝成一個(gè) Request 對(duì)象、然后傳遞到視圖函數(shù)中。
但對(duì)于 flask 而言則不是這樣子的,我們看一下 flask 是如何接收請(qǐng)求參數(shù)的。
from flask import Flask, request app = Flask("flask") @app.route("/index") def index(): name = request.args.get("name") return {"name": name} app.run(host="127.0.0.1", port=7777)
我們看到對(duì)于 flask 而言則是通過 import request 的方式,如果不需要的話就不用 import,當(dāng)然我這里并不是在比較哪種方式好,主要是為了引出我們今天的主題。首先對(duì)于 flask 而言,如果我再定義一個(gè)視圖函數(shù)的話,那么獲取請(qǐng)求參數(shù)依舊是相同的方式,但是這樣問題就來了,不同的視圖函數(shù)內(nèi)部使用同一個(gè) request,難道不會(huì)發(fā)生沖突嗎?
顯然根據(jù)我們使用 flask 的經(jīng)驗(yàn)來說,答案是不會(huì)的,至于原因就是 ThreadLocal。
ThreadLocal,從名字上看可以得出它肯定是和線程相關(guān)的。沒錯(cuò),它專門用來創(chuàng)建局部變量,并且創(chuàng)建的局部變量是和線程綁定的。
import threading # 創(chuàng)建一個(gè) local 對(duì)象 local = threading.local() def get(): name = threading.current_thread().name # 獲取綁定在 local 上的 value value = local.value print(f"線程: {name}, value: {value}") def set_(): name = threading.current_thread().name # 為不同的線程設(shè)置不同的值 if name == "one": local.value = "ONE" elif name == "two": local.value = "TWO" # 執(zhí)行 get 函數(shù) get() t1 = threading.Thread(target=set_, name="one") t2 = threading.Thread(target=set_, name="two") t1.start() t2.start() """ 線程 one, value: ONE 線程 two, value: TWO """
可以看到兩個(gè)線程之間是互不影響的,因?yàn)槊總€(gè)線程都有自己唯一的 id,在綁定值的時(shí)候會(huì)綁定在當(dāng)前的線程中,獲取也會(huì)從當(dāng)前的線程中獲取??梢园?ThreadLocal 想象成一個(gè)字典:
{ "one": {"value": "ONE"}, "two": {"value": "TWO"} }
更準(zhǔn)確的說 key 應(yīng)該是線程的 id,為了直觀我們就用線程的 name 代替了,但總之在獲取的時(shí)候只會(huì)獲取綁定在該線程上的變量的值。
而 flask 內(nèi)部也是這么設(shè)計(jì)的,只不過它沒有直接用 threading.local,而是自己實(shí)現(xiàn)了一個(gè) Local 類,除了支持線程之外還支持 greenlet 的協(xié)程,那么它是怎么實(shí)現(xiàn)的呢?首先我們知道 flask 內(nèi)部存在 "請(qǐng)求 context" 和 "應(yīng)用 context",它們都是通過棧來維護(hù)的(兩個(gè)不同的棧)。
# flask/globals.py _request_ctx_stack = LocalStack() _app_ctx_stack = LocalStack() current_app = LocalProxy(_find_app) request = LocalProxy(partial(_lookup_req_object, "request")) session = LocalProxy(partial(_lookup_req_object, "session"))
每個(gè)請(qǐng)求都會(huì)綁定在當(dāng)前的 Context 中,等到請(qǐng)求結(jié)束之后再銷毀,這個(gè)過程由框架完成,開發(fā)者只需要直接使用 request 即可。所以請(qǐng)求的具體細(xì)節(jié)流程可以點(diǎn)進(jìn)源碼中查看,這里我們重點(diǎn)關(guān)注一個(gè)對(duì)象:werkzeug.local.Local,也就是上面說的 Local 類,它是變量的設(shè)置和獲取的關(guān)鍵。直接看部分源碼:
# werkzeug/local.py class Local(object): __slots__ = ("__storage__", "__ident_func__") def __init__(self): # 內(nèi)部有兩個(gè)成員:__storage__ 是一個(gè)字典,值就存在這里面 # __ident_func__ 只需要知道它是用來獲取線程 id 的即可 object.__setattr__(self, "__storage__", {}) object.__setattr__(self, "__ident_func__", get_ident) def __call__(self, proxy): """Create a proxy for a name.""" return LocalProxy(self, proxy) def __release_local__(self): self.__storage__.pop(self.__ident_func__(), None) def __getattr__(self, name): try: # 根據(jù)線程 id 得到 value(一個(gè)字典) # 然后再根據(jù) name 獲取對(duì)應(yīng)的值 # 所以只會(huì)獲取綁定在當(dāng)前線程上的值 return self.__storage__[self.__ident_func__()][name] except KeyError: raise AttributeError(name) def __setattr__(self, name, value): ident = self.__ident_func__() storage = self.__storage__ try: # 將線程 id 作為 key,然后將值設(shè)置在對(duì)應(yīng)的字典中 # 所以只會(huì)將值設(shè)置在當(dāng)前的線程中 storage[ident][name] = value except KeyError: storage[ident] = {name: value} def __delattr__(self, name): # 刪除邏輯也很簡(jiǎn)單 try: del self.__storage__[self.__ident_func__()][name] except KeyError: raise AttributeError(name)
所以我們看到 flask 內(nèi)部的邏輯其實(shí)很簡(jiǎn)單,通過 ThreadLocal 實(shí)現(xiàn)了線程之間的隔離。每個(gè)請(qǐng)求都會(huì)綁定在各自的 Context 中,獲取值的時(shí)候也會(huì)從各自的 Context 中獲取,因?yàn)樗褪怯脕肀4嫦嚓P(guān)信息的(重要的是同時(shí)也實(shí)現(xiàn)了隔離)。
相應(yīng)此刻你已經(jīng)理解了上下文,但是問題來了,不管是 threading.local 也好、還是類似于 flask 自己實(shí)現(xiàn)的 Local 也罷,它們都是針對(duì)線程的。如果是使用 async def 定義的協(xié)程該怎么辦呢?如何實(shí)現(xiàn)每個(gè)協(xié)程的上下文隔離呢?所以終于引出了我們的主角:contextvars。
該模塊提供了一組接口,可用于在協(xié)程中管理、設(shè)置、訪問局部 Context 的狀態(tài)。
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試") async def get(): # 獲取值 return c.get() + "~~~" async def set_(val): # 設(shè)置值 c.set(val) print(await get()) async def main(): coro1 = set_("協(xié)程1") coro2 = set_("協(xié)程2") await asyncio.gather(coro1, coro2) asyncio.run(main()) """ 協(xié)程1~~~ 協(xié)程2~~~ """
ContextVar 提供了兩個(gè)方法,分別是 get 和 set,用于獲取值和設(shè)置值。我們看到效果和 ThreadingLocal 類似,數(shù)據(jù)在協(xié)程之間是隔離的,不會(huì)受到彼此的影響。
但我們?cè)僮屑?xì)觀察一下,我們是在 set_ 函數(shù)中設(shè)置的值,然后在 get 函數(shù)中獲取值???await get() 相當(dāng)于是開啟了一個(gè)新的協(xié)程,那么意味著設(shè)置值和獲取值不是在同一個(gè)協(xié)程當(dāng)中。但即便如此,我們依舊可以獲取到希望的結(jié)果。因?yàn)?Python 的協(xié)程是無棧協(xié)程,通過 await 可以實(shí)現(xiàn)級(jí)聯(lián)調(diào)用。
我們不妨再套一層:
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試") async def get1(): return await get2() async def get2(): return c.get() + "~~~" async def set_(val): # 設(shè)置值 c.set(val) print(await get1()) print(await get2()) async def main(): coro1 = set_("協(xié)程1") coro2 = set_("協(xié)程2") await asyncio.gather(coro1, coro2) asyncio.run(main()) """ 協(xié)程1~~~ 協(xié)程1~~~ 協(xié)程2~~~ 協(xié)程2~~~ """
我們看到不管是 await get1() 還是 await get2(),得到的都是 set_ 中設(shè)置的結(jié)果,說明它是可以嵌套的。
并且在這個(gè)過程當(dāng)中,可以重新設(shè)置值。
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試") async def get1(): c.set("重新設(shè)置") return await get2() async def get2(): return c.get() + "~~~" async def set_(val): # 設(shè)置值 c.set(val) print("------------") print(await get2()) print(await get1()) print(await get2()) print("------------") async def main(): coro1 = set_("協(xié)程1") coro2 = set_("協(xié)程2") await asyncio.gather(coro1, coro2) asyncio.run(main()) """ ------------ 協(xié)程1~~~ 重新設(shè)置~~~ 重新設(shè)置~~~ ------------ ------------ 協(xié)程2~~~ 重新設(shè)置~~~ 重新設(shè)置~~~ ------------ """
先 await get2() 得到的就是 set_ 函數(shù)中設(shè)置的值,這是符合預(yù)期的。但是我們?cè)?get1 中將值重新設(shè)置了,那么之后不管是 await get1() 還是直接 await get2(),得到的都是新設(shè)置的值。
這也說明了,一個(gè)協(xié)程內(nèi)部 await 另一個(gè)協(xié)程,另一個(gè)協(xié)程內(nèi)部 await 另另一個(gè)協(xié)程,不管套娃(await)多少次,它們獲取的值都是一樣的。并且在任意一個(gè)協(xié)程內(nèi)部都可以重新設(shè)置值,然后獲取會(huì)得到最后一次設(shè)置的值。再舉個(gè)栗子:
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試") async def get1(): return await get2() async def get2(): val = c.get() + "~~~" c.set("重新設(shè)置啦") return val async def set_(val): # 設(shè)置值 c.set(val) print(await get1()) print(c.get()) async def main(): coro = set_("古明地覺") await coro asyncio.run(main()) """ 古明地覺~~~ 重新設(shè)置啦 """
await get1() 的時(shí)候會(huì)執(zhí)行 await get2(),然后在里面拿到 c.set 設(shè)置的值,打印 "古明地覺~~~"。但是在 get2 里面,又將值重新設(shè)置了,所以第二個(gè) print 打印的就是新設(shè)置的值。\
如果在 get 之前沒有先 set,那么會(huì)拋出一個(gè) LookupError,所以 ContextVar 支持默認(rèn)值:
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試", default="哼哼") async def set_(val): print(c.get()) c.set(val) print(c.get()) async def main(): coro = set_("古明地覺") await coro asyncio.run(main()) """ 哼哼 古明地覺 """
除了在 ContextVar 中指定默認(rèn)值之外,也可以在 get 中指定:
import asyncio import contextvars c = contextvars.ContextVar("只是一個(gè)標(biāo)識(shí), 用于調(diào)試", default="哼哼") async def set_(val): print(c.get("古明地戀")) c.set(val) print(c.get()) async def main(): coro = set_("古明地覺") await coro asyncio.run(main()) """ 古明地戀 古明地覺 """
所以結(jié)論如下,如果在 c.set 之前使用 c.get:
當(dāng) ContextVar 和 get 中都沒有指定默認(rèn)值,會(huì)拋出 LookupError;
只要有一方設(shè)置了,那么會(huì)得到默認(rèn)值;
如果都設(shè)置了,那么以 get 為準(zhǔn);
如果 c.get 之前執(zhí)行了 c.set,那么無論 ContextVar 和 get 有沒有指定默認(rèn)值,獲取到的都是 c.set 設(shè)置的值。
所以總的來說還是比較好理解的,并且 ContextVar 除了可以作用在協(xié)程上面,它也可以用在線程上面。沒錯(cuò),它可以替代 threading.local,我們來試一下:
import threading import contextvars c = contextvars.ContextVar("context_var") def get(): name = threading.current_thread().name value = c.get() print(f"線程 {name}, value: {value}") def set_(): name = threading.current_thread().name if name == "one": c.set("ONE") elif name == "two": c.set("TWO") get() t1 = threading.Thread(target=set_, name="one") t2 = threading.Thread(target=set_, name="two") t1.start() t2.start() """ 線程 one, value: ONE 線程 two, value: TWO """
和 threading.local 的表現(xiàn)是一樣的,但是更建議使用 ContextVars。不過前者可以綁定任意多個(gè)值,而后者只能綁定一個(gè)值(可以通過傳遞字典的方式解決這一點(diǎn))。
當(dāng)我們調(diào)用 c.set 的時(shí)候,其實(shí)會(huì)返回一個(gè) Token 對(duì)象:
import contextvars c = contextvars.ContextVar("context_var") token = c.set("val") print(token) """ <Token var=<ContextVar name='context_var' at 0x00..> at 0x00...> """
Token 對(duì)象有一個(gè) var 屬性,它是只讀的,會(huì)返回指向此 token 的 ContextVar 對(duì)象。
import contextvars c = contextvars.ContextVar("context_var") token = c.set("val") print(token.var is c) # True print(token.var.get()) # val print( token.var.set("val2").var.set("val3").var is c ) # True print(c.get()) # val3
Token 對(duì)象還有一個(gè) old_value 屬性,它會(huì)返回上一次 set 設(shè)置的值,如果是第一次 set,那么會(huì)返回一個(gè) <Token.MISSING>。
import contextvars c = contextvars.ContextVar("context_var") token = c.set("val") # 該 token 是第一次 c.set 所返回的 # 在此之前沒有 set,所以 old_value 是 <Token.MISSING> print(token.old_value) # <Token.MISSING> token = c.set("val2") print(c.get()) # val2 # 返回上一次 set 的值 print(token.old_value) # val
那么這個(gè) Token 對(duì)象有什么作用呢?從目前來看貌似沒太大用處啊,其實(shí)它最大的用處就是和 reset 搭配使用,可以對(duì)狀態(tài)進(jìn)行重置。
import contextvars #### c = contextvars.ContextVar("context_var") token = c.set("val") # 顯然是可以獲取的 print(c.get()) # val # 將其重置為 token 之前的狀態(tài) # 但這個(gè) token 是第一次 set 返回的 # 那么之前就相當(dāng)于沒有 set 了 c.reset(token) try: c.get() # 此時(shí)就會(huì)報(bào)錯(cuò) except LookupError: print("報(bào)錯(cuò)啦") # 報(bào)錯(cuò)啦 # 但是我們可以指定默認(rèn)值 print(c.get("默認(rèn)值")) # 默認(rèn)值
它負(fù)責(zé)保存 ContextVars 對(duì)象和設(shè)置的值之間的映射,但是我們不會(huì)直接通過 contextvars.Context 來創(chuàng)建,而是通過 contentvars.copy_context 函數(shù)來創(chuàng)建。
import contextvars c1 = contextvars.ContextVar("context_var1") c1.set("val1") c2 = contextvars.ContextVar("context_var2") c2.set("val2") # 此時(shí)得到的是所有 ContextVar 對(duì)象和設(shè)置的值之間的映射 # 它實(shí)現(xiàn)了 collections.abc.Mapping 接口 # 因此我們可以像操作字典一樣操作它 context = contextvars.copy_context() # key 就是對(duì)應(yīng)的 ContextVar 對(duì)象,value 就是設(shè)置的值 print(context[c1]) # val1 print(context[c2]) # val2 for ctx, value in context.items(): print(ctx.get(), ctx.name, value) """ val1 context_var1 val1 val2 context_var2 val2 """ print(len(context)) # 2
除此之外,context 還有一個(gè) run 方法:
import contextvars c1 = contextvars.ContextVar("context_var1") c1.set("val1") c2 = contextvars.ContextVar("context_var2") c2.set("val2") context = contextvars.copy_context() def change(val1, val2): c1.set(val1) c2.set(val2) print(c1.get(), context[c1]) print(c2.get(), context[c2]) # 在 change 函數(shù)內(nèi)部,重新設(shè)置值 # 然后里面打印的也是新設(shè)置的值 context.run(change, "VAL1", "VAL2") """ VAL1 VAL1 VAL2 VAL2 """ print(c1.get(), context[c1]) print(c2.get(), context[c2]) """ val1 VAL1 val2 VAL2 """
我們看到 run 方法接收一個(gè) callable,如果在里面修改了 ContextVar 實(shí)例設(shè)置的值,那么對(duì)于 ContextVar 而言只會(huì)在函數(shù)內(nèi)部生效,一旦出了函數(shù),那么還是原來的值。但是對(duì)于 Context 而言,它是會(huì)受到影響的,即便出了函數(shù),也是新設(shè)置的值,因?yàn)樗苯影褍?nèi)部的字典給修改了。
關(guān)于“Python怎么利用contextvars實(shí)現(xiàn)管理上下文變量”這篇文章的內(nèi)容就介紹到這里,感謝各位的閱讀!相信大家對(duì)“Python怎么利用contextvars實(shí)現(xiàn)管理上下文變量”知識(shí)都有一定的了解,大家如果還想學(xué)習(xí)更多知識(shí),歡迎關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。