版权声明:转载请标明出处 https://blog.csdn.net/gymaisyl/article/details/86712940
class AppContext(object):
"""应用程序上下文将应用程序对象隐式绑定到当前线程或greenlet,
类似于:class:`RequestContext`绑定请求信息。
如果创建了请求上下文但应用程序不是,则也会隐式创建应用程序上下文 在个人应用程序上下文之上。
"""
def __init__(self, app):
self.app = app
self.url_adapter = app.create_url_adapter(None)
self.g = app.app_ctx_globals_class()
# 与请求上下文一样,应用程序上下文可以多次推送,但基本的“引用计数”足以跟踪它们。
self._refcnt = 0
def push(self):
"""将应用程序上下文绑定到当前上下文。"""
self._refcnt += 1
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
_app_ctx_stack.push(self)
appcontext_pushed.send(self.app)
def pop(self, exc=_sentinel):
"""删除应用程序上下文。"""
try:
self._refcnt -= 1
if self._refcnt <= 0:
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_appcontext(exc)
finally:
rv = _app_ctx_stack.pop()
assert rv is self, 'Popped wrong app context. (%r instead of %r)' \
% (rv, self)
appcontext_popped.send(self.app)
def __enter__(self):
self.push()
return self
def __exit__(self, exc_type, exc_value, tb):
self.pop(exc_value)
if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
reraise(exc_type, exc_value, tb)
class RequestContext(object):
"""
请求上下文包含所有与请求相关的信息。它是在请求开始时创建的,
并被推送到“_request_ctx_stack”并在请求结束时删除。它将为提供的WSGI环境创建URL适配器和请求对象。
不要尝试直接使用这个类,而是使用
:meth:“~ flask.Flask。test_request_context”和
:meth:“~ flask.Flask。request_context '来创建这个对象。
当请求上下文弹出时,它将计算所有
在应用程序上注册的用于拆卸执行的函数
(:meth:“~ flask.Flask.teardown_request”)。
请求上下文会在请求的末尾自动弹出。在调试模式下,如果发生异常,请求上下文将保留,以便交互式调试器有机会内省数据。
对于没有失败且在“调试”模式之外的请求,也可以强制使用0.4。
通过设置“meth。在WSGI环境中,上下文不会在请求结束时弹出自己。
这是由:meth:' ~flask.Flask。例如test_client '来实现延迟清理功能。
您可能会发现这对单元测试很有帮助,因为您需要
从上下文本地获取信息的时间更长一些。确保正确:meth:' ~werkzeug.LocalStack。在这种情况下自己弹出堆栈,
否则单元测试将泄漏内存。
"""
def __init__(self, app, environ, request=None):
self.app = app
if request is None:
request = app.request_class(environ)
self.request = request
self.url_adapter = app.create_url_adapter(self.request)
self.flashes = None
self.session = None
# 可以多次推送请求上下文并与其他请求上下文交织。
# 现在,只有弹出最后一个级别,我们才能摆脱它们。
# 此外,如果缺少应用程序上下文,则会隐式创建,因此对于每个级别,我们都会添加此信息
self._implicit_app_ctx_stack = []
# 如果保留上下文,则指示。 下次推送另一个上下文时,将弹出保留的上下文。
self.preserved = False
# 如果存在上下文保留的情况,则会记住pop的异常。
self._preserved_exc = None
# 在响应对象上发出请求后应执行的函数。 这些将在常规“after_request”函数之前调用。
self._after_request_functions = []
self.match_request()
def _get_g(self):
return _app_ctx_stack.top.g
def _set_g(self, value):
_app_ctx_stack.top.g = value
g = property(_get_g, _set_g)
del _get_g, _set_g
def copy(self):
"""
使用相同的请求对象创建此请求上下文的副本。
这可用于将请求上下文移动到不同的greenlet。
因为实际的请求对象是相同的,所以除非锁定对请求对象的访问,否则不能将请求上下文移动到不同的线程。
"""
return self.__class__(self.app,
environ=self.request.environ,
request=self.request
)
def match_request(self):
"""
可以被子类覆盖,以连接到请求的匹配。
"""
try:
url_rule, self.request.view_args = \
self.url_adapter.match(return_rule=True)
self.request.url_rule = url_rule
except HTTPException as e:
self.request.routing_exception = e
def push(self):
"""将请求上下文绑定到当前上下文。"""
"""
如果在调试模式下发生异常,或者在异常情况下激活了上下文保留,则只有一个上下文保留在堆栈上。
理由是您希望在调试情况下访问该信息。 但是,如果有人忘记再次弹出该上下文,
我们希望确保在下一次推送时它无效,否则我们会冒一些泄漏内存的风险。
这通常只是测试套件中的一个问题,因为此功能在生产环境中不活动。
"""
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
# 在我们推送请求上下文之前,我们必须确保存在应用程序上下文。
app_ctx = _app_ctx_stack.top
if app_ctx is None or app_ctx.app != self.app:
app_ctx = self.app.app_context()
app_ctx.push()
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
_request_ctx_stack.push(self)
# 在请求上下文可用时打开会话。 这允许自定义open_session方法使用请求上下文。
# 如果这是第一次推送请求,则仅打开新会话,否则stream_with_context将丢失会话。
if self.session is None:
session_interface = self.app.session_interface
self.session = session_interface.open_session(
self.app, self.request
)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
def pop(self, exc=_sentinel):
"""
弹出请求上下文并通过执行此操作解除绑定。 这也将触发执行注册的功能
:meth:`~flask.Flask.teardown_request`装饰。
"""
app_ctx = self._implicit_app_ctx_stack.pop()
try:
clear_request = False
if not self._implicit_app_ctx_stack:
self.preserved = False
self._preserved_exc = None
if exc is _sentinel:
exc = sys.exc_info()[1]
self.app.do_teardown_request(exc)
# 如果此解释器支持清除异常信息,我们现在就这样做。 这只会在Python 2.x上生效,
# 在3.x上,它在异常堆栈结束时自动消失。
if hasattr(sys, 'exc_clear'):
sys.exc_clear()
request_close = getattr(self.request, 'close', None)
if request_close is not None:
request_close()
clear_request = True
finally:
rv = _request_ctx_stack.pop()
# 在请求结束时摆脱循环依赖关系,这样我们就不需要GC处于活动状态。
if clear_request:
rv.request.environ['werkzeug.request'] = None
# 如有必要,摆脱应用程序
if app_ctx is not None:
app_ctx.pop(exc)
assert rv is self, 'Popped wrong request context. ' \
'(%r instead of %r)' % (rv, self)
def auto_pop(self, exc):
if self.request.environ.get('flask._preserve_context') or \
(exc is not None and self.app.preserve_context_on_exception):
self.preserved = True
self._preserved_exc = exc
else:
self.pop(exc)
def __enter__(self):
self.push()
return self
def __exit__(self, exc_type, exc_value, tb):
# 如果我们处于调试模式并且发生异常,请不要弹出请求堆栈。
# 这将允许调试器仍然访问交互式shell中的请求对象。 此外上下文可以为测试客户端强制保持活动状态。
self.auto_pop(exc_value)
if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
reraise(exc_type, exc_value, tb)
def __repr__(self):
return '<%s \'%s\' [%s] of %s>' % (
self.__class__.__name__,
self.request.url,
self.request.method,
self.app.name,
)