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参数。
参考链接
- dormousehole.readthedocs.io/en/latest/
- python-main stackoverflow.com/questions/4…
- click click.palletsprojects.com/en/8.0.x/
- dotenv pypi.org/project/pyt…
