给寻找编程代码教程的朋友们精选了相关的编程文章，网友国信鸥根据主题投稿了本篇教程内容，涉及到Flask、上下文、Flask核心机制之上下文源码剖析相关内容，已被846网友关注，相关难点技巧可以阅读下方的电子资料。
Flask核心机制之上下文源码剖析
一、前言
了解过flask的python开发者想必都知道flask中核心机制莫过于上下文管理，当然学习flask如果不了解其中的处理流程，可能在很多问题上不能得到解决，当然我在写本篇文章之前也看到了很多博文有关于对flask上下文管理的剖析都非常到位，当然为了学习flask我也把对flask上下文理解写下来供自己参考，也希望对其他人有所帮助。
二、知识储备
threadlocal
在多线程中，线程间的数据是共享的， 但是每个线程想要有自己的数据该怎么实现？ python中的threading.local对象已经实现，其原理是利用线程的唯一标识作为key，数据作为value来保存其自己的数据，以下是demo演示了多个线程同时修改同一变量的值的结果：
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # Author:wd import threading import time values=threading.local() def run(arg): values.num=arg #修改threading.local对象的name数据 time.sleep(1) print(threading.current_thread().name,values.num) #打印values.num for i in range(3): th = threading.Thread(target=run, args=(i,), name='run thread%s' % i) th.start()
结果：
run thread0 0
run thread1 1
run thread2 2
结果说明：
从结果中可以看到，values.num的值是不同的，按照普通线程理解因为有sleep存在，在每个线程最后打印values.num时候值应该都是2，但是正是因为threading.local对象内部会为每个线程开辟一个内存空间，从而使得每个线程都有自己的单独数据，所以每个线程修改的是自己的数据(内部实现为字典)，打印结果才不一样。
有了以上的设计思想，我们可以自己定义类似于thread.local类，为了支持协程，将其唯一标识改为协程的唯一标识，其实这已经及其接近flask中的Local类了(后续在进行说明)：
try:
from greenlet import getcurrent as get_ident # 携程唯一标识
except ImportError:
try:
from thread import get_ident
except ImportError:
from _thread import get_ident # 线程唯一标识
class Local(object):
def __init__(self):
object.__setattr__(self, 'storage', dict()) # 防止self.xxx 递归
object.__setattr__(self, '__get_ident__', get_ident)
def __setattr__(self, key, value):
ident = self.__get_ident__() # 获取当前线程或协程的唯一标识
data = self.storage.get(ident)
if not data: # 当前线程没有数据
data = {key: value} # 创建数据
else: # 当前已经有数据
data[key] = value
self.storage[ident] = data # 最后为当前线程设置其标识对应的数据
def __getattr__(self, name):
try:
return self.storage[self.__get_ident__()].get(name) # 返回name所对应的值
except KeyError:
raise AttributeError(name)
functools.partial
partial函数是工具包的一个不常用函数，其作用是给函数传递参数，同时返回的也是这个函数，但是这个函数的已经带了参数了，示例：
from functools import partial def func(x,y,z): print(x,y,z) new_fun=partial(func,1,2) #生成新的函数，该函数中已经有一个参数 new_fun(3)
结果：
1 2 3
在以上示例中，new_func是由func生成的，它已经参数1,2了，只需要传递3即可运行。
werkzeug
werkzeug是一个实现了wsgi协议的模块，用官方语言介绍：Werkzeug is a WSGI utility library for Python. It's widely used and BSD licensed。为什么会提到它呢，这是因为flask内部使用的wsgi模块就是werkzeug，以下是一个示例(如果你了解wsgi协议的应该不用过多介绍)：
from werkzeug.wrappers import Request, Response
@Request.application
def application(request):
return Response('Hello World!')
if __name__ == '__main__':
from werkzeug.serving import run_simple
run_simple('localhost', 4000, application)
在示例中application是一个可调用的对象也可以是带有__call__方法的对象，在run_simple内部执行application(),也就是在源码的execute(self.server.app)中执行，这里你只需要run_simple会执行第三个参数加括号。
三、源码剖析
上下文管理
在说请求上下文之前先看一个flask的hell world示例：
from flask import Flask
app=Flask(__name__)
@app.route("/")
def hello():
return 'hello world'
if __name__=='__main__':
app.run()
在以上示例中，app.run是请求的入口，而app是Flask实例化的对象，所以执行的是Flask类中的run方法，而在该改方法中又执行了run_simple方法，以下是run方法部分源码摘抄(其中self就是app对象)：
from werkzeug.serving import run_simple try: run_simple(host, port, self, **options) finally: # reset the first request information if the development server # reset normally. This makes it possible to restart the server # without reloader and that stuff from an interactive shell. self._got_first_request = False
在run_simple中会执行app(environ, start_response),参考werkzeug的源码,源码会执行app(environ, start_response)也就是执行app的__call__方法，以下是__call__方法源码摘抄：
def __call__(self, environ, start_response): """The WSGI server calls the Flask application object as the WSGI application. This calls :meth:`wsgi_app` which can be wrapped to applying middleware.""" return self.wsgi_app(environ, start_response)
__call__方法中又调用了wsgi_app方法,该方法也就是flask的核心所在，下面是方法摘抄：
def wsgi_app(self, environ, start_response):
"""The actual WSGI application. This is not implemented in
:meth:`__call__` so that middlewares can be applied without
losing a reference to the app object. Instead of doing this::
app = MyMiddleware(app)
It's a better idea to do this instead::
app.wsgi_app = MyMiddleware(app.wsgi_app)
Then you still have the original application object around and
can continue to call methods on it.
.. versionchanged:: 0.7
Teardown events for the request and app contexts are called
even if an unhandled error occurs. Other events may not be
called depending on when an error occurs during dispatch.
See :ref:`callbacks-and-errors`.
:param environ: A WSGI environment.
:param start_response: A callable accepting a status code,
a list of headers, and an optional exception context to
start the response.
"""
#ctx.app 当前app名称
#ctx.request request对象，由app.request_class(environ)生成
#ctx.session session 相关信息
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
#push数据到local，此时push的数据分请求上线文和应用上下文
# 将ctx通过Localstack添加到local中
# app_ctx是APPContext对象
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)
第一句：ctx = self.request_context(environ)调用request_context实例化RequestContext对象,以下是RequestContext类的构造方法：
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
此时的request为None，所以self.request=app.request_class(environ)，而在Flask类中request_class = Request，此时执行的是Request(environ),也就是实例化Request类，用于封装请求数据，最后返回RequestContext对象，此时的ctx含有以下属性ctx.app(app对象)、ctx.request(请求封装的所有请求信息)、ctx.app(当前app对象)等
第二句：ctx.push()， 调用RequestContext的push方法，以下是源码摘抄：
def push(self):
"""Binds the request context to the current context."""
# If an exception occurs in debug mode or if context preservation is
# activated under exception situations exactly one context stays
# on the stack. The rationale is that you want to access that
# information under debug situations. However if someone forgets to
# pop that context again we want to make sure that on the next push
# it's invalidated, otherwise we run at risk that something leaks
# memory. This is usually only a problem in test suite since this
# functionality is not active in production environments.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop(top._preserved_exc)
# Before we push the request context we have to ensure that there
# is an application context.
app_ctx = _app_ctx_stack.top #获取应用上线文，一开始为none
if app_ctx is None or app_ctx.app != self.app:
# 创建APPContext(self)对象，app_ctx=APPContext(self)
# 包含app_ctx.app ,当前app对象
# 包含app_ctx.g , g可以看作是一个字典用来保存一个请求周期需要保存的值
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()
#self 是RequestContext对象，其中包含了请求相关的所有数据
_request_ctx_stack.push(self)
# Open the session at the moment that the request context is available.
# This allows a custom open_session method to use the request context.
# Only open a new session if this is the first time the request was
# pushed, otherwise stream_with_context loses the session.
if self.session is None:
session_interface = self.app.session_interface # 获取session信息
self.session = session_interface.open_session(
self.app, self.request
)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
到了这里可以看到，相关注解已经标注，flask内部将上下文分为了app_ctx(应用上下文)和_request_ctx(请求上下文)，并分别用来两个LocalStack()来存放各自的数据(以下会用request_ctx说明，当然app_ctx也一样)，其中app_ctx包含app、url_adapter一下是app_ctx构造方法：
def __init__(self, app): self.app = app self.url_adapter = app.create_url_adapter(None) self.g = app.app_ctx_globals_class() # Like request context, app contexts can be pushed multiple times # but there a basic "refcount" is enough to track them. self._refcnt = 0
然后分别执行app_ctx.push()方法和_request_ctx_stack.push(self)方法，将数据push到stack上,_request_ctx_stack.push(self)，而_request_ctx_stack是一个LocalStack对象，是一个全局对象，具体路径在flask.globals,以下是其push方法：
def push(self, obj):
"""Pushes a new item to the stack"""
#找_local对象中是否有stack，没有设置rv和_local.stack都为[]
rv = getattr(self._local, 'stack', None)
if rv is None:
self._local.stack = rv = []
# 执行Local对象的__setattr__方法，等价于a=[],rv=a, self._local.stack =a
#创建字典，类似于storage={'唯一标识':{'stack':[]}}
rv.append(obj)
#列表中追加请求相关所有数据也就是storage={'唯一标识':{'stack':[RequestContext对象,]}}
return rv
以上代码中的self._local是一个Local()对象源码定义如下,也就是用于存储每次请求的数据，和我们刚开始定义的local及其相似，这也是为什么要先提及下threadlocal。
Local()
class Local(object):
__slots__ = ('__storage__', '__ident_func__')
def __init__(self):
object.__setattr__(self, '__storage__', {})
object.__setattr__(self, '__ident_func__', get_ident)
def __iter__(self):
return iter(self.__storage__.items())
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:
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:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
Local()
到这里我们知道了，当执行ctx.push()时，local对象中已经有数据了,接着开始执行self.full_dispatch_request()，也就是开始执行视图函数，以下是源码摘抄：
def full_dispatch_request(self):
"""Dispatches the request and on top of that performs request
pre and postprocessing as well as HTTP exception catching and
error handling.
.. versionadded:: 0.7
"""
self.try_trigger_before_first_request_functions()
try:
request_started.send(self)
rv = self.preprocess_request()
if rv is None:
rv = self.dispatch_request()
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
在改方法中调用self.preprocess_request(),用于执行所有被before_request装饰器装饰的函数，从源码总可以看到如果该函数有返回，则不会执行self.dispatch_request()也就是视图函数，
执行完毕之后调用self.dispatch_request()根据路由匹配执行视图函数,然后响应最后调用ctx.auto_pop(error)将stack中的数据删除，此时完成一次请求。
全局对象request、g、session
在了解完flask的上下文管理时候，我们在视图函数中使用的request实际上是一个全局变量对象，当然还有g、session这里以request为例子，它是一个LocalProxy对象，以下是源码片段：
request = LocalProxy(partial(_lookup_req_object, 'request'))
当我们使用request.path时候实际上是调用是其__getattr__方法即LocalProxy对象的__getattr__方法，我们先来看看LocalProxy对象实例化的参数：
def __init__(self, local, name=None):
#local是传入的函数，该句等价于self.__local=local,_类名__字段强行设置私有字段值
#如果是requst则函数就是partial(_lookup_req_object, 'request')
object.__setattr__(self, '_LocalProxy__local', local)
object.__setattr__(self, '__name__', name) #开始的时候设置__name__的值为None
if callable(local) and not hasattr(local, '__release_local__'):
# "local" is a callable that is not an instance of Local or
# LocalManager: mark it as a wrapped function.
object.__setattr__(self, '__wrapped__', local)
在源码中实例化时候传递的是partial(_lookup_req_object, 'request')函数作为参数，也就是self.__local=该函数，partial参数也就是我们之前提到的partial函数，作用是传递参数，此时为_lookup_req_object函数传递request参数，这个在看看其__getattr__方法：
def __getattr__(self, name):
#以获取request.method 为例子，此时name=method
if name == '__members__':
return dir(self._get_current_object())
#self._get_current_object()返回的是ctx.request,再从ctx.request获取method (ctx.request.method)
return getattr(self._get_current_object(), name)
在以上方法中会调用self._get_current_object()方法，而_get_current_object()方法中会调用self.__local()也就是带参数request参数的 _lookup_req_object方法从而返回ctx.request(请求上下文)，最后通过然后反射获取name属性的值，这里我们name属性是path，如果是request.method name属性就是method，最后我们在看看_lookup_req_object怎么获取到的ctx.request，以下是源码摘抄:
def _lookup_req_object(name):
#以name=request为列
top = _request_ctx_stack.top
# top是就是RequestContext(ctx)对象，里面含有request、session 等
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name) #到RequestContext（ctx）中获取那么为request的值
在源码中很简单无非就是利用_request_ctx_stack(也就是LocalStack对象)的top属性返回stack中的ctx,在通过反射获取request，最后返回ctx.request。以上是整个flask的上下文核心机制，与其相似的全局对象有如下(session、g)：
# context locals _request_ctx_stack = LocalStack() #LocalStack()包含pop、push方法以及Local对象，上下文通过该对象push和pop _app_ctx_stack = LocalStack() current_app = LocalProxy(_find_app) request = LocalProxy(partial(_lookup_req_object, 'request')) #reuqest是LocalProxy的对象，设置和获取request对象中的属性通过LocalProxy定义的各种双下划线实现 session = LocalProxy(partial(_lookup_req_object, 'session')) g = LocalProxy(partial(_lookup_app_object, 'g'))
技巧应用
利用flask的上下文处理机制我们获取上请求信息还可以使用如下方式:
from flask import Flask,_request_ctx_stack
app=Flask(__name__)
@app.route("/")
def hello():
print(_request_ctx_stack.top.request.method) #结果GET,等价于request.method
return 'this is wd'
if __name__=='__main__':
app.run()
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持码农之家。
