前言

Flask是目前为止我最喜欢的一个Python Web框架了，为了更好的掌握其内部实现机制，这两天准备学习下Flask的源码，将由浅入深跟大家分享下，其中Flask版本为1.1.1。

上次了解了Flask服务的启动流程，今天我们来看下路由的内部实现机理。

Flask系列文章：

1. Flask开发初探
2. Flask源码分析一：服务启动

关于路由

所谓路由，就是处理请求URL和函数之间关系的程序。

Flask中也是对URL规则进行统一管理的，创建URL规则有两种方式：

1. 使用@app.route修饰器，并传入URL规则作为参数，将函数绑定到URL，这个过程便将一个函数注册为路由，这个函数则被称为视图函数。
2. 使用app.add_url_rule()。

在开始阅读源码之前，我是有这几点疑问的？

那就让我们带着这几点疑问一起去学习源码吧！

正文

注册路由

首先，route()装饰器：

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

route()有两个参数，rule表示url规则。该函数对参数进行处理之后，调用方法add_url_role()，这里也就验证了两种注册路由的方法等价。我们来看下代码：

def add_url_rule(
self,
rule,
endpoint=None,
view_func=None,
provide_automatic_options=None,
**options
):

if endpoint is None:
endpoint = _endpoint_from_view_func(view_func)
options["endpoint"] = endpoint
methods = options.pop("methods", None)

# if the methods are not given and the view_func object knows its
# methods we can use that instead.  If neither exists, we go with
# a tuple of only ``GET`` as default.
if methods is None:
methods = getattr(view_func, "methods", None) or ("GET",)
if isinstance(methods, string_types):
raise TypeError(
"Allowed methods have to be iterables of strings, "
'for example: @app.route(..., methods=["POST"])'
)
methods = set(item.upper() for item in methods)

# Methods that should always be added
required_methods = set(getattr(view_func, "required_methods", ()))

# starting with Flask 0.8 the view_func object can disable and
# force-enable the automatic options handling.
if provide_automatic_options is None:
provide_automatic_options = getattr(
view_func, "provide_automatic_options", None
)

if provide_automatic_options is None:
if "OPTIONS" not in methods:
provide_automatic_options = True
required_methods.add("OPTIONS")
else:
provide_automatic_options = False

# Add the required methods now.
methods |= required_methods

rule = self.url_rule_class(rule, methods=methods, **options)
rule.provide_automatic_options = provide_automatic_options

self.url_map.add(rule)
if view_func is not None:
old_func = self.view_functions.get(endpoint)
if old_func is not None and old_func != view_func:
raise AssertionError(
"View function mapping is overwriting an "
"existing endpoint function: %s" % endpoint
)
self.view_functions[endpoint] = view_func

入参包括：

1. rule: url规则
2. endpoint : 要注册规则的endpoint，默认是视图函数的名儿
3. view_func: 视图函数
4. provide_automatic_options: 请求方法是否添加OPTIONS方法的一个标志
5. options: 关于请求处理的一些方法等

可以看到，add_url_rule()首先进行参数处理，包括：

1. endpoint默认为视图函数的name
2. url请求的方法默认为GET
3. 若请求方法中没有设置OPTIONS，添加该方法。

在处理完所有的参数后，将该URL规则写入url_map（创建好Rule对象，并添加到Map对象中），将视图函数写入view_function字典中。

其中，url_map 是

werkzeug.routing:Map

werkzeug.routing:Rule

werkzeug

werkzeug是使用Python编写的一个WSGI工具集，werkzeug.routing模块主要用于url解析。

Rule类

Rule类继承自RuleFactory类，一个Rule实例代表一个URL模式，一个WSGI应用会处理很多个不同的URL模式，与此同时产生很多个Rule实例，这些实例将作为参数传给Map类。

Map类

Map类构造的实例存储所有的url规则，解析并匹配请求对应的视图函数。

路由匹配

在应用初始化的过程中，会注册所有的路由规则，可以调用（app.url_map）查看，当服务收到URL请求时，就需要进行路由匹配，以找到对应的视图函数，对应的流程和原理是什么呢？

当用户请求进入Flask应用时，调用Flask类的wsgi_app方法:

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:
error = e
response = self.handle_exception(e)
except:  # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)

该函数的处理过程包括：

1. 创建RequestContext对象，在对象初始化的过程中调用app.create_url_adapter()方法，将请求参数environ传给Map对象创建MapAdapter对象，保存在url_adapter字段中
2. 将RequestContext对象推入_request_ctx_stack栈中
3. 通过RequestContext的match_request方法，调用MapAdapter对象的match方法找到匹配的Rule并解析出参数，保存在request的url_rule和view_args字段中
4. 调用full_dispatch_request()

接下来我们看下full_dispatch_request方法：

def full_dispatch_request(self):
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)

可以看到，重点执行dispatch_request()：

def dispatch_request(self):
req = _request_ctx_stack.top.request
if req.routing_exception is not None:
self.raise_routing_exception(req)
rule = req.url_rule
# if we provide automatic options for this URL and the
# request came with the OPTIONS method, reply automatically
if (
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
return self.view_functions[rule.endpoint](**req.view_args)

处理的过程是：获取请求对象的request，找到对应的endpoint，继而从view_functions中找到对应的视图函数，传递请求参数，视图函数处理内部逻辑并返回，完成一次请求分发。

以上，就是Flask路由的内部实现原理。