达州做网站
大橙子网站建设，新征程启航
为企业提供网站建设、域名注册、服务器等服务
flask上下文
flask上下文预备知识点
面向对象attr
class Foo(object):
def __init__(self):
# self.storage = {}
object.__setattr__(self, 'storage', {})
def __setattr__(self, key, value):
self.storage[key] = value
def __getattr__(self, item):
return self.storage.get(item)
cls = Foo() #创建对象
cls.x1 = 123 #调用setattr方法
cls.x1 #调用getattr方法
线程的唯一标识
import threading
from threading import get_ident
def task():
ident = get_ident() # 线程的唯一标识 线程id
print(ident)
for i in range(10):
t = threading.Thread(target=task)
t.start()
local对象与flask的上下文
每一个request都是一个线程，每个线程有自己唯一的线程ID，利用这个线程ID作为标识，来存储request对象。
这样的话就可以为每个线程开辟独立的空间互不影响。
创新互联专注于邕宁网站建设服务及定制，我们拥有丰富的企业做网站经验。 热诚为您提供邕宁营销型网站建设，邕宁网站制作、邕宁网页设计、邕宁网站官网定制、重庆小程序开发公司服务，打造邕宁网络公司原创品牌,更为您提供邕宁网站排名全网营销落地服务。
自己实现threading.local
点击查看代码
import threading
from threading import get_ident
'''
storage = {
1111:{x1:1}
1112:{x1:2}
1113:{x1:3}
1114:{x1:4}
}
'''
class Local(object):
def __init__(self):
object.__setattr__(self, 'storage', {}) # 相当于self.storage = {}
def __setattr__(self, key, value):
ident = get_ident()
if ident in self.storage: # 如果ident在字典里面就直接修改就行了
self.storage[ident][key] = value
else: # 不在字典里面就添加
self.storage[ident] = {key:value}
def __getattr__(self,item):
ident = get_ident()
if ident not in self.storage:
return
return self.storage[ident].get(item)
local = Local()
def task(num):
local.x1 = num
print(local.x1)
if __name__ == '__main__':
for i in range(5):
t = threading.Thread(target=task, args=(i,))
t.start()
加强版threading.local
点击查看代码
import threading
from threading import get_ident
'''
storage = {
1111:{x1:[1]}
1112:{x1:[]}
1113:{x1:[]}
1114:{x1:[]}
}
维护成一个栈
'''
class Local(object):
def __init__(self):
object.__setattr__(self, 'storage', {}) # 相当于self.storage = {}
def __setattr__(self, key, value):
ident = get_ident()
if ident in self.storage: # 如果ident在字典里面就直接修改就行了
self.storage[ident][key].append(value)
else: # 不在字典里面就添加
self.storage[ident] = {key:[value]}
def __getattr__(self,item):
ident = get_ident()
if ident not in self.storage:
return
return self.storage[ident][item][-1]
local = Local()
def task(num):
local.x1 = num
print(local.x1)
if __name__ == '__main__':
for i in range(5):
t = threading.Thread(target=task, args=(i,))
t.start()
flask源码的local对象和localstack对象
flask的locak对象
点击查看代码
class Local:
__slots__ = ("_storage",)
def __init__(self) -> None:
object.__setattr__(self, "_storage", ContextVar("local_storage"))
@property
def __storage__(self) -> t.Dict[str, t.Any]:
warnings.warn(
"'__storage__' is deprecated and will be removed in Werkzeug 2.1.",
DeprecationWarning,
stacklevel=2,
)
return self._storage.get({}) # type: ignore
@property
def __ident_func__(self) -> t.Callable[[], int]:
warnings.warn(
"'__ident_func__' is deprecated and will be removed in"
" Werkzeug 2.1. It should not be used in Python 3.7+.",
DeprecationWarning,
stacklevel=2,
)
return _get_ident # type: ignore
@__ident_func__.setter
def __ident_func__(self, func: t.Callable[[], int]) -> None:
warnings.warn(
"'__ident_func__' is deprecated and will be removed in"
" Werkzeug 2.1. Setting it no longer has any effect.",
DeprecationWarning,
stacklevel=2,
)
def __release_local__(self) -> None:
__release_local__(self._storage)
def __getattr__(self, name: str) -> t.Any:
values = self._storage.get({})
try:
return values[name]
except KeyError:
raise AttributeError(name) from None
def __setattr__(self, name: str, value: t.Any) -> None:
values = self._storage.get({}).copy()
values[name] = value
self._storage.set(values)
def __delattr__(self, name: str) -> None:
values = self._storage.get({}).copy()
try:
del values[name]
self._storage.set(values)
except KeyError:
raise AttributeError(name) from None
flask的localstack对象
点击查看代码
class LocalStack:
def __init__(self) -> None:
self._local = Local()
def __release_local__(self) -> None:
self._local.__release_local__()
@property
def __ident_func__(self) -> t.Callable[[], int]:
return self._local.__ident_func__
@__ident_func__.setter
def __ident_func__(self, value: t.Callable[[], int]) -> None:
object.__setattr__(self._local, "__ident_func__", value)
def __call__(self) -> "LocalProxy":
def _lookup() -> t.Any:
rv = self.top
if rv is None:
raise RuntimeError("object unbound")
return rv
return LocalProxy(_lookup)
def push(self, obj: t.Any) -> t.List[t.Any]:
"""Pushes a new item to the stack"""
rv = getattr(self._local, "stack", []).copy()
rv.append(obj)
self._local.stack = rv
return rv # type: ignore
def pop(self) -> t.Any:
"""Removes the topmost item from the stack, will return the
old value or `None` if the stack was already empty.
"""
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
release_local(self._local)
return stack[-1]
else:
return stack.pop()
@property
def top(self) -> t.Any:
"""The topmost item on the stack. If the stack is empty,
`None` is returned.
"""
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
# 总结
在flask中有个local类，他和threading.local的功能一样，为每个线程开辟空间来存储数据，他们两个的内部实现机制一样，内部维护一个字典，以线程（协程）ID为key，进行数据隔离如：
__storage__ = {
1111:{'k1':123}
}
obj = local()
obj.k1 = 123
在flask中还有localstack的类，他内部会依赖local对象，local对象负责存储数据，localstack对象用于将local中的值维护成一个栈。
__storage__ = {
1112:{'stack':[k1,]}
}
obj = LocalStack()
obj.push('k1')
obj.pop()
obj.top
flask源码中一共有两个localstack对象
context locals
__storage__ = {
'1111':{'stack':['RequestContext(request, session)',]},
'1112':{'stack':['RequestContext(request, session)',]}
}
__storage__ = {
'1111':{'stack':['AppContext(app, g)',]},
'1112':{'stack':['AppContext(app, g)',]}
}
_request_ctx_stack = LocalStack() # 请求上下文
_app_ctx_stack = LocalStack() # 应用上下文
新闻名称：flask上下文
文章路径：http://dzwzjz.com/article/dsojssp.html