Flask 源码阅读：正菜
The following article is from 游戏不存在 Author 肖恩
这是上篇：《Flask 源码阅读：开胃菜》
flask项目大名鼎鼎，不需要多做介绍。我把它称之为python服务开发的TOP2项目，另外一个就是django。这两个项目各有千秋，各自有不同的应用场景，都需要深入理解，熟练掌握。本次源码选择的版本是 1.1.2，我会采用慢读法，尽自己最大努力把它讲透。本篇是第2篇-正菜，主要包括:
flask-app创建 Requestcontext && AppContext的实现 request && response 的处理 小结 小技巧
flask-app创建
还是先从flask-app的示例开始:
from flask import Flask
app = Flask(__name__)
@app.route("/")
def hello_world():
return "<p>Hello, World!</p>"
app的启动，在「flask源码阅读：开胃菜」中有过介绍，这里就不再赘述。我们在示例中创建了一个flask-app，然后给这个app注册了一个路由处理函数，返回一段html。跟随这个示例，我们查看flask-app的创建:
class Flask(_PackageBoundObject):
response_class = Response
request_class = Request
app_ctx_globals_class = _AppCtxGlobals
config_class = Config
def __init__(
self,
import_name,
static_url_path=None,
static_folder="static",
static_host=None,
host_matching=False,
subdomain_matching=False,
template_folder="templates",
instance_path=None,
instance_relative_config=False,
root_path=None,
):
# import_name使用__name__
_PackageBoundObject.__init__(
self, import_name, template_folder=template_folder, root_path=root_path
)
...
主要有下面三点：
Flask类继承自_PackageBoundObject类 response_class、request_class、app_ctx_globals_class和config_class 这几个重要的类都以策略模式提供，方便扩展 定义默认的静态文件目录 static 和默认的模板目录 templates
_PackageBoundObject类主要实现当前app根目录获取:
class _PackageBoundObject(object):
import_name = None
template_folder = None
root_path = None
def __init__(self, import_name, template_folder=None, root_path=None):
self.import_name = import_name
self.template_folder = template_folder
if root_path is None:
root_path = get_root_path(self.import_name)
self.root_path = root_path
self._static_folder = None
...
获取根目录后，默认static中的静态文件和templates中的模板文件就可以提供读取，这样flask-app不需要额外使用werkzeug中的StaticMiddleware。比如静态文件目录的提供:
@property
def static_folder(self):
"""The absolute path to the configured static folder."""
if self._static_folder is not None:
return os.path.join(self.root_path, self._static_folder)
flask-app中实现wsgi-application规范:
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
error = None
try:
try:
ctx.push()
response = self.full_dispatch_request()
except Exception as e:
...
return response(environ, start_response)
finally:
...
ctx.auto_pop(error)
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response)
创建RequestContext(ctx)对象, 注意这里没有直接处理Request对象,而是使用ctx.push 通过full_dispatch_request处理请求获取到response对象 返回response 使用auto-pop释放RequestContext对象(ctx)
RequestContext && AppContext
Request上下文(RequestContext)是flask中最重要的类，下面是它的构造函数:
class RequestContext(object):
def __init__(self, app, environ, request=None, session=None):
self.app = app
if request is None:
request = app.request_class(environ)
self.request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request)
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session
self._implicit_app_ctx_stack = []
...
引用flask-app，利用app的request_class创建真正的request对象 create_url_adapter 主要使用werkzeug的router进行wsgi-environ的处理
push方法中，创建app-context然后入栈，同时创建session:
def push(self):
...
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()
# 加入request-ctx的本地栈
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
_request_ctx_stack.push(self)
if self.session is None:
session_interface = self.app.session_interface
self.session = session_interface.open_session(self.app, self.request)
...
if self.url_adapter is not None:
self.match_request()
match_request是对request的处理，稍后介绍Request部分再详细介绍。_app_ctx_stack和_request_ctx_stack是两个个非常重要的概念，现在我们只需要记住这是2个栈即可，先后push进去app-ctx和request-ctx，以后我会单独介绍这2个概念。
RequestContext的pop方法进行出栈操作:
def pop(self, exc=_sentinel):
# 清理本地栈
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)
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()
# get rid of circular dependencies at the end of the request
# so that we don't require the GC to be active.
if clear_request:
rv.request.environ["werkzeug.request"] = None
# Get rid of the app as well if necessary.
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):
...
self.pop(exc)
和入栈相反，先进行request_ctx_的出栈，再进行app_ctx的出栈。
AppContext包装App和RequestContext包装Request类似:
class AppContext(object):
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
def push(self):
"""Binds the app context to the current context."""
self._refcnt += 1
...
_app_ctx_stack.push(self)
# appcontext_pushed是一个信号，在介绍blink库时候已经有过介绍
appcontext_pushed.send(self.app)
def pop(self, exc=_sentinel):
"""Pops the app context."""
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 也是信号
appcontext_popped.send(self.app)
可以看到这里的push和pop与之前的RequestContext基本一致。
request && response
request解析的第1步是从match_request开始，使用url_adapter分离出http协议的url(url_rule)和query参数(view_args):
# RequestContext
def match_request(self):
try:
result = self.url_adapter.match(return_rule=True)
self.request.url_rule, self.request.view_args = result
except HTTPException as e:
...
第2步是使用full_dispatch_request，进行分发前的预处理:
# Flask
def full_dispatch_request(self):
# 触发request请求的前置函数(和信号类似)
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:
...
return self.finalize_request(rv)
第3步是使用dispatch_request将request的分发请求到路由:
# Flask
def dispatch_request(self):
# 直接从栈中获取request对象
req = _request_ctx_stack.top.request
...
rule = req.url_rule
...
return self.view_functions[rule.endpoint](**req.view_args)
view_functions是通过app的router装饰器注册而来:
# Flask
@setupmethod
def add_url_rule(
self,
rule,
endpoint=None,
view_func=None,
provide_automatic_options=None,
**options
):
...
self.url_map.add(rule)
if view_func is not None:
...
self.view_functions[endpoint] = view_func
def route(self, rule, **options):
def decorator(f):
endpoint = options.pop("endpoint", None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decorator
request经过上述主要的3步后，业务完成，开始进入response的处理环节。先是生成response:
# Flask
def finalize_request(self, rv, from_error_handler=False):
response = self.make_response(rv)
try:
response = self.process_response(response)
# 又是信号
request_finished.send(self, response=response)
except Exception:
...
return response
response处理第1步就是构建header的status:
def make_response(self, rv):
status = headers = None
# unpack tuple returns
if isinstance(rv, tuple):
len_rv = len(rv)
# a 3-tuple is unpacked directly
if len_rv == 3:
rv, status, headers = rv
# decide if a 2-tuple has status or headers
elif len_rv == 2:
if isinstance(rv[1], (Headers, dict, tuple, list)):
rv, headers = rv
else:
rv, status = rv
...
# 处理body
rv = self.response_class(rv, status=status, headers=headers)
status = headers = None
...
return rv
response处理第2步就是将response进行额外处理，类似request的预处理一样，比如包装上session。
def process_response(self, response):
ctx = _request_ctx_stack.top
bp = ctx.request.blueprint
funcs = ctx._after_request_functions
if bp is not None and bp in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[bp]))
if None in self.after_request_funcs:
funcs = chain(funcs, reversed(self.after_request_funcs[None]))
for handler in funcs:
response = handler(response)
if not self.session_interface.is_null_session(ctx.session):
self.session_interface.save_session(self, ctx.session, response)
return response
小结
Flask实现了wsgi-applicaction的规范，对于每一个请求。Flask都会创建RequestCtx和AppCtx对象，保证在请求的生命周期中都可以很方便的获取Request对象和App对象。而对于每个请求，分成request和response两个环节，这也是我们在werkzeug中介绍过的 洋葱模型 。Flask对request的处理分:请求URL分析，request的预处理和路由派发3步；response的处理经过状态码和head/body封装，response后置处理器2步。
以上就是Flask的核心流程了，希望这道正菜大家满意，可以给一个赞或者在看。
小技巧
使用_request_ctx_stack这种全局堆栈，可以减少参数的传递. 观测下面派发request到路由的函数:
# Flask
def dispatch_request(self):
# 直接从栈中获取request对象
req = _request_ctx_stack.top.request
...
rule = req.url_rule
...
return self.view_functions[rule.endpoint](**req.view_args)
这里dispatch_request函数没有request参数，这样避免了Flask-App对request处理的每个函数都要附带上request参数。
参考链接
https://dormousehole.readthedocs.io/en/latest/ python-main https://stackoverflow.com/questions/4042905/what-is-main-py click https://click.palletsprojects.com/en/8.0.x/ dotenv https://pypi.org/project/python-dotenv/
ps:本文为该系列文章第二篇，关注【Python开发者】公众号，锁定后续更新。
- EOF -
觉得本文对你有帮助？请分享给更多人
推荐关注「Python开发者」，提升Python技能
点赞和在看就是最大的支持❤️