Flask源码篇：Flask路由规则与请求匹配过程

1 启动时路由相关操作

所谓的路由原理，就是Flask如何创建自己的路由体系，并当一个请求到来时，如何根据路由体系准确定位处理函数，并响应请求。

本节还是以最简单一个Flask应用例子来展开讲解路由原理，如下：

from flask import Flask

app = Flask(__name__)@app.route('/')defhello_world():return'Hello World!'if __name__ =='__main__':
    app.run()

（1）分析app.route()

首先路由的注册是通过scaffold(Flask继承scaffold)包下的

route()

装饰器实现的，其源码如下：

defroute(self, rule:str,**options: t.Any)-> t.Callable:defdecorator(f: t.Callable)-> t.Callable:# 获取endpoint
        endpoint = options.pop("endpoint",None)# 添加路由，rule就是app.route()传来的路由字符串，及'/'
        self.add_url_rule(rule, endpoint, f,**options)return f

    return decorator

可以看到这个装饰器主要做了两件事：1.获取

endpoint

；2.添加路由。

**其中最重要的是函数

add_url_rule()

，用来添加路由映射。**

补充：

endpoint是在后面Flask存储路由和函数名映射的时候用到，如果没有指定，默认是被装饰的函数名。具体如何使用，后面还会分析到；
因为app.route()的本质还是add_url_rule()函数，所以我们也可以直接使用这个函数，用法可以参考文章Flask路由。

（2）分析add_url_rule()

下面来看下add_url_rule做了哪些事，其核心源码如下：

classFlask(Scaffold):# 这里只有要讨论的主要代码，其他代码省略了
    
    url_rule_class = Rule
    url_map_class = Map
    
    def__init__(
        self,
        import_name:str,
        static_url_path: t.Optional[str]=None,
        static_folder: t.Optional[t.Union[str, os.PathLike]]="static",
        static_host: t.Optional[str]=None,
        host_matching:bool=False,
        subdomain_matching:bool=False,
        template_folder: t.Optional[str]="templates",
        instance_path: t.Optional[str]=None,
        instance_relative_config:bool=False,
        root_path: t.Optional[str]=None,):super().__init__(
            import_name=import_name,
            static_folder=static_folder,
            static_url_path=static_url_path,
            template_folder=template_folder,
            root_path=root_path,)
        
        self.url_map = self.url_map_class()
        self.url_map.host_matching = host_matching
        self.subdomain_matching = subdomain_matching
  
    defadd_url_rule(
        self,
        rule:str,
        endpoint: t.Optional[str]=None,
        view_func: t.Optional[t.Callable]=None,
        provide_automatic_options: t.Optional[bool]=None,**options: t.Any,)->None:# 1.如果没提供endpoint，获取默认的endpointif endpoint isNone:
             endpoint = _endpoint_from_view_func(view_func)# type: ignore
         options["endpoint"]= endpoint

          # 2.获取请求方法，在装饰器@app.route('/index', methods=['POST','GET'])的method参数# 如果没有指定，则给个默认的元组("GET",)# 关于provide_automatic_options处理一些暂不看
          methods = options.pop("methods",None)if methods isNone:
              methods =getattr(view_func,"methods",None)or("GET",)ifisinstance(methods,str):raise TypeError("Allowed methods must be a list of strings, for"' example: @app.route(..., methods=["POST"])')
          methods ={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 isNone:
              provide_automatic_options =getattr(
                  view_func,"provide_automatic_options",None)if provide_automatic_options isNone:if"OPTIONS"notin methods:
                  provide_automatic_options =True
                  required_methods.add("OPTIONS")else:
                  provide_automatic_options =False# Add the required methods now.
          methods |= required_methods

          # 3.重要的一步：url_rule_class方法实例化Rule对象
          rule = self.url_rule_class(rule, methods=methods,**options)
          rule.provide_automatic_options = provide_automatic_options  # type: ignore# 4.重要的一步：url_map（Map对象）的add方法
          self.url_map.add(rule)# 5.判断endpoint和view_func的映射存不存在，如果已经有其他view_func用了这个endpoint，则报错，否则新的映射加到self.view_functions里# self.view_functions继承自Scaffold，是一个字典对象if view_func isnotNone:
              old_func = self.view_functions.get(endpoint)if old_func isnotNoneand old_func != view_func:raise AssertionError("View function mapping is overwriting an existing"f" endpoint function: {endpoint}")
              self.view_functions[endpoint]= view_func

分析上面源码，这个方法主要做了以下几件事：

如果没提供endpoint，获取默认的endpoint；
获取请求方法，在装饰器@app.route('/index', methods=['POST','GET'])的method参数，如果没有指定，则给个默认的元组(“GET”,)；
self.url_rule_class()：实例化Rule对象，Rule类后面还会再讲解；
调用self.url_map.add()方法，其中self.url_map是一个Map对象，后面还会再讲解；
self.view_functions[endpoint] = view_func添加endpoint和view_func的映射。

其中实例化

Rule

对象和

self.url_map.add()

是Falsk路由里的核心，下面分析

Rule

类和

Map

类。

（3）分析Rule类

Rule

类在

werkzeug.routing

模块下，其源码较多，这里也只摘取我们用到的主要代码，如下：

classRule(RuleFactory):def__init__(
        self,
        string:str,
        methods: t.Optional[t.Iterable[str]]=None,
        endpoint: t.Optional[str]=None,# 此处省略了其他参数的代码 ...)->None:ifnot string.startswith("/"):raise ValueError("urls must start with a leading slash")
        self.rule = string
        self.is_leaf =not string.endswith("/")
        self.map:"Map"=None# type: ignore
        self.methods = methods
        self.endpoint:str= endpoint 
                # 省略了其他初始化的代码defget_rules(self,map:"Map")-> t.Iterator["Rule"]:"""获取map对象的rule迭代器"""yield self
    
    defbind(self,map:"Map", rebind:bool=False)->None:"""把map对象绑定到Rule对象上，并且根据rule和map信息创建一个path正则表达式，存储在rule对象的self._regex属性里，路由匹配的时候用"""if self.mapisnotNoneandnot rebind:raise RuntimeError(f"url rule {self!r} already bound to map {self.map!r}")# 把map对象绑定到Rule对象上
        self.map=mapif self.strict_slashes isNone:
            self.strict_slashes =map.strict_slashes
        if self.merge_slashes isNone:
            self.merge_slashes =map.merge_slashes
        if self.subdomain isNone:
            self.subdomain =map.default_subdomain
        # 调用compile方法创建一个正则表达式
        self.compile()defcompile(self)->None:"""编写正则表达式并存储到属性self._regex中"""# 此处省略了正则的解析过程代码
        regex =f"^{''.join(regex_parts)}{tail}$"
        self._regex = re.compile(regex)defmatch(
        self, path:str, method: t.Optional[str]=None)-> t.Optional[t.MutableMapping[str, t.Any]]:"""这个函数用于校验传进来path参数（路由）是否能够匹配，匹配不上返回None"""# 省去了部分代码，只摘录了主要代码，看一下大致逻辑即可ifnot self.build_only:
            require_redirect =False# 1.根据bind后的正则结果（self._regex正则）去找path的结果集
            m = self._regex.search(path)if m isnotNone:
                groups = m.groupdict()# 2.编辑匹配到的结果集，加到一个result字典里并返回
            result ={}for name, value in groups.items():try:
                      value = self._converters[name].to_python(value)except ValidationError:returnNone
                result[str(name)]= value
                if self.defaults:
                      result.update(self.defaults)return result
        returnNone

Rule

类继承自

RuleFactory

类，主要参数有：

string：路由字符串
methods：路由方法
endpoint：endpoint参数

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

（4）分析Map类

Map

类也在werkzeug.routing模块下，其源码较多，这里也只摘取我们用到的主要代码，主要源码如下：

classMap:def__init__(
        self,
        rules: t.Optional[t.Iterable[RuleFactory]]=None# 此处省略了其他参数)->None:# 根据传进来的rules参数维护了一个私有变量self._rules列表
        self._rules: t.List[Rule]=[]# endpoint和rule的映射
        self._rules_by_endpoint: t.Dict[str, t.List[Rule]]={}# 此处省略了其他初始化操作defadd(self, rulefactory: RuleFactory)->None:"""
        把Rule对象或一个RuleFactory对象添加到map并且绑定到map，要求rule没被绑定过
        """for rule in rulefactory.get_rules(self):# 调用rule对象的bind方法
            rule.bind(self)# 把rule对象添加到self._rules列表里
            self._rules.append(rule)# 把endpoint和rule的映射加到属性self._rules_by_endpoint里
            self._rules_by_endpoint.setdefault(rule.endpoint,[]).append(rule)
        self._remap =Truedefbind(
        self,
        server_name:str,
        script_name: t.Optional[str]=None,
        subdomain: t.Optional[str]=None,
        url_scheme:str="http",
        default_method:str="GET",
        path_info: t.Optional[str]=None,
        query_args: t.Optional[t.Union[t.Mapping[str, t.Any],str]]=None,)->"MapAdapter":"""
        返回一个新的类MapAdapter
        """
        server_name = server_name.lower()if self.host_matching:if subdomain isnotNone:raise RuntimeError("host matching enabled and a subdomain was provided")elif subdomain isNone:
            subdomain = self.default_subdomain
        if script_name isNone:
            script_name ="/"if path_info isNone:
            path_info ="/"try:
            server_name = _encode_idna(server_name)# type: ignoreexcept UnicodeError as e:raise BadHost()from e

        return MapAdapter(
            self,
            server_name,
            script_name,
            subdomain,
            url_scheme,
            path_info,
            default_method,
            query_args,)

Map

类有两个个非常重要的属性：

self._rules，属性是一个列表，存储了一系列Rule对象；
self._rules_by_endpoint：

其中有个核心方法

add()

，这里就是我们分析

app.add_url_rule()

方法是第4步调用的方法。后面会详细讲解。

（5）分析MapAdapter类

在

Map

类中，会用到

MapAdapter

类，下面我们认识下这个类：

classMapAdapter:"""`Map.bind`或`Map.bind_to_environ` 会返回这个类
    主要用来做匹配
    """defmatch(
        self,
        path_info: t.Optional[str]=None,
        method: t.Optional[str]=None,
        return_rule:bool=False,
        query_args: t.Optional[t.Union[t.Mapping[str, t.Any],str]]=None,
        websocket: t.Optional[bool]=None,)-> t.Tuple[t.Union[str, Rule], t.Mapping[str, t.Any]]:"""匹配请求的路由和Rule对象"""# 只摘摘录了主要代码，省略了大量代码...# 这里是主要步骤：遍历map对象的rule列表，依次和path进行匹配for rule in self.map._rules:try:# 调用rule对象的match方法返回匹配结果
                rv = rule.match(path, method)except RequestPath as e:# 下面省略了大量代码...# 返回rule对象（或endpoint）和匹配的路由结果if return_rule:return rule, rv
         else:return rule.endpoint, rv

Map.bind

或

Map.bind_to_environ

方法会返回

MapAdapter

对象。**

**其中

MapAdapter

对象核心方法是

match

，主要步骤是遍历map对象的rule列表，依次和path进行匹配，当然也是调用rule对象的match方法返回匹配结果。**

接下来可以稍微独立的看下

Map

类和

Rlue

对象联合起来怎么用，看下面一个例子：

from werkzeug.routing import Map, Rule

m = Map([
    Rule('/', endpoint='index'),
    Rule('/blog', endpoint='blog/index'),
    Rule('/blog/<int:id>', endpoint='blog/detail')])# 返回一个MapAdapter对象
map_adapter = m.bind("example.com","/")# MapAdapter对象的 match方法会返回匹配的结果print(map_adapter.match("/","GET"))# ('index', {})print(map_adapter.match("/blog/42"))# ('blog/detail', {'id': 42})print(map_adapter.match("/blog"))# ('blog/index', {})

可以看到，

Map

对象通过

bind

返回了一个

MapAdapter

对象，

MapAdapter

对象的

match

方法可以找到路由匹配的结果。

（6）分析 url_rule_class()

add_url_rule

的第一个主要步骤就是

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

，即创建一个

Rule

对象。

在分析

Rlue

类的时候，知道了Rule对象主要有

string

（路由字符串）、

methods

、

endpoint

3个属性。在下面以一个具体的示例，看下实例化的

Rule

对象是什么样子的。

还是一开始最上面的示例，我们看下通过

@app.route('/')

代码后到实例化Rule对象的时候，这个对象的具体属性，通过debug如下:

在这里插入图片描述

可以看到，rule对象的

rule

属性是传的路由，

endpoint

属性是通过函数名获取的，

method

属性是支持的请求方法，'HEAD’和

OPTIONS

是默认添加的。

（7）分析map.add(rule)

add_url_rule

的第二个主要步骤就是

self.url_map.add(rule)

，即调用

Map

对象的add方法。

在第4步分析Map对象时，有提到过，现在再回头仔细看下这个方法做了什么：

defadd(self, rulefactory: RuleFactory)->None:"""
        把Rule对象或一个RuleFactory对象添加到map并且绑定到map，要求rule没被绑定过
        """for rule in rulefactory.get_rules(self):# 调用rule对象的bind方法
            rule.bind(self)# 把rule对象添加到self._rules列表里
            self._rules.append(rule)# 把endpoint和rule的映射加到属性self._rules_by_endpoint里
            self._rules_by_endpoint.setdefault(rule.endpoint,[]).append(rule)
        self._remap =True

其实主要就是把上一步实例化的Rule对象或一个RuleFactory对象添加到Map对象并且绑定到map。

主要做了两件事：

调用rule对象的bind方法rule.bind(self)：在分析Rule类时，提到过这个方法，它主要作用是把map对象绑定到Rule对象上，并且根据rule和map信息创建一个正则表达式（Rule对象的self._regex 属性）。
把rule对象添加到Map对象的self._rules列表里；
把endpoint和rule的映射加到Map对象的属性self._rules_by_endpoint（一个字典）里；

我们可以通过示例看下Map对象在add之后变成了什么样，通过debug，结果如下：

在这里插入图片描述

可以看到，

Map

的

self._rules

和

self._rules_by_endpoint

属性都含有新加的’/'路由对应的数据（其中

/static/<path:filename>

是默认添加的静态文件的位置路由）。

上面就分析完了，如何添加一个路由映射的。

2 请求进来时路由匹配过程

（1）分析wsgi_app

下面分析下，当前一个请求进来时，如何根据前面

Map

和

Rlue

对象来进行路由匹配的。

上一章中我们分析了当请求进来时会先经过wsgi服务器处理完请求后调用Flask app的

__call__

方法，其代码如下：

def__call__(self, environ:dict, start_response: t.Callable)-> t.Any:"""The WSGI server calls the Flask application object as the
    WSGI application. This calls :meth:`wsgi_app`, which can be
    wrapped to apply middleware.
    """return self.wsgi_app(environ, start_response)

可以看到其实就是调用了app的

wsgi_app()

方法。其代码如下：

defwsgi_app(self, environ:dict, start_response: t.Callable)-> t.Any:# 1.获取请求上下文
    ctx = self.request_context(environ)
    error: t.Optional[BaseException]=Nonetry:try:# 2.调用请求上下文的push方法
            ctx.push()# 3.调用full_dispatch_request()分发请求，获取响应结果
            response = self.full_dispatch_request()except Exception as e:
            error = e
            response = self.handle_exception(e)except: 
            error = sys.exc_info()[1]raisereturn response(environ, start_response)finally:if self.should_ignore_error(error):
            error =None
        ctx.auto_pop(error)

其中有3个主要步骤：

获取请求上下文：ctx = self.request_context(environ)
调用请求上下文的push方法：ctx.push()
调用full_dispatch_request()分发请求，获取响应结果：response = self.full_dispatch_request()

下面来逐个分析下每个步骤的作用。

（2）分析request_context

wsgi_app

方法的第一个主要步骤。

这个方法主要就是获取一个上下文对象。

需要把

environ

（环境变量等）传入到方法中。上下文在Flask中也是一个重要的概念，当然关于上下文的解析下章会重点解析。此章只关注我们我们需要的。

defrequest_context(self, environ:dict)-> RequestContext:return RequestContext(self, environ)

此方法就是创建一个

RequestContext

对象。

RequestContext

类部分源码如下：

classRequestContext:def__init__(
        self,
        app:"Flask",
        environ:dict,
        request: t.Optional["Request"]=None,
        session: t.Optional["SessionMixin"]=None,)->None:
        self.app = app
        if request isNone:
            request = app.request_class(environ)
        self.request = request
        self.url_adapter =Nonetry:# 此处是重点，调用了Falsk对象的create_url_adapter方法获取了MapAdapter对象
            self.url_adapter = app.create_url_adapter(self.request)except HTTPException as e:
            self.request.routing_exception = e
        self.flashes =None
        self.session = session
                # 其他代码省略...# 其他方法的源码省略...

**在创建

RequestContext

对象的初始化方法中，有个很重要的步骤就是获取

MapAdapter

对象**。

前面第4节我们分析过它的作用，主要就是用来匹配路由的。

下面我们看下

create_url_adapter

源码：

defcreate_url_adapter(
    self, request: t.Optional[Request])-> t.Optional[MapAdapter]:if request isnotNone:ifnot self.subdomain_matching:
            subdomain = self.url_map.default_subdomain orNoneelse:
            subdomain =None# 此处是重点，调用了Map对象的bind_to_environ方法return self.url_map.bind_to_environ(
            request.environ,
            server_name=self.config["SERVER_NAME"],
            subdomain=subdomain,)if self.config["SERVER_NAME"]isnotNone:# 此处是重点，调用了Map对象的bind方法return self.url_map.bind(
            self.config["SERVER_NAME"],
            script_name=self.config["APPLICATION_ROOT"],
            url_scheme=self.config["PREFERRED_URL_SCHEME"],)returnNone

**可以看到，这个方法的主要作用就是调用

Map

对象的

bind_to_environ

方法或

bind

方法。前面讲Map类的时候也分析过，这两个方法主要是返回了

MapAdapter

对象。**

（3）分析ctx.push

wsgi_app

方法的第二个主要步骤。

wsgi方法里获取到上下文对象后，就调用了

push

方法，其代码如下（只保留了核心代码）：

classRequestContext:def__init__(
        self,
        app:"Flask",
        environ:dict,
        request: t.Optional["Request"]=None,
        session: t.Optional["SessionMixin"]=None,)->None:# 代码省略passdefmatch_request(self)->None:try:# 1.调用了MapAdapter对象的match方法，返回了rule对象和参数对象
            result = self.url_adapter.match(return_rule=True)# 2.把rule对象和参数对象放到请求上下文中
            self.request.url_rule, self.request.view_args = result
        except HTTPException as e:
            self.request.routing_exception = e

    defpush(self)->None:"""Binds the request context to the current context."""# 此处省略了前置校验处理代码（上下文、session等处理）if self.url_adapter isnotNone:# 调用了match_request方法
            self.match_request()

可以看到

push

方法主要是调用了

match_request

方法，这个方法主要做了如下两件事：

调用了MapAdapter对象的match方法，会根据Map对象存储的路由信息去匹配当前请求的路由，返回了rule对象和参数对象。
把rule对象和参数对象放到请求上下文中。

（4）分析full_dispatch_request

wsgi_app

方法的第三个主要步骤。

full_dispatch_request

方法源码如下：

deffull_dispatch_request(self)-> Response:
    self.try_trigger_before_first_request_functions()try:
        request_started.send(self)
        rv = self.preprocess_request()if rv isNone:# 这里是主要的步骤：分发请求
            rv = self.dispatch_request()except Exception as e:
        rv = self.handle_user_exception(e)return self.finalize_request(rv)

**其中

dispatch_request()

是核心方法。**

dispatch_request()

方法源码如下：

defdispatch_request(self)-> ResponseReturnValue:# 1.获取请求上下文对象
    req = _request_ctx_stack.top.request
    if req.routing_exception isnotNone:
        self.raise_routing_exception(req)# 2.从上下文中获取前面存在里面的Rule对象
    rule = req.url_rule
    # if we provide automatic options for this URL and the# request came with the OPTIONS method, reply automaticallyif(getattr(rule,"provide_automatic_options",False)and req.method =="OPTIONS"):return self.make_default_options_response()# 这里是重点：根据rule对象的endpoint属性从self.view_functions属性中获取对应的视图函数，# 然后把上下文中的参数传到视图函数中并调用视图函数处理请求，返回处理结果return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)

这个方法主要步骤如下：

req = _request_ctx_stack.top.request：获取请求上下文对象
rule = req.url_rule：从上下文中获取前面存在里面的Rule对象，就是ctx.push()方法放进上下文的
根据rule对象的endpoint属性从self.view_functions属性中获取对应的视图函数，然后把上下文中的参数传到视图函数中并调用视图函数处理请求，返回处理结果：return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)

至此，一个完整的请求就处理完啦。

3 总结

根据以上分析，对路由规则和请求匹配总结如下：

应用启动时：

app.route()调用add_url_rule方法。
add_url_rule方法里：调用self.url_rule_class()实例化Rule对象；
add_url_rule方法里：调用self.url_map.add()方法，把Rule对象、endpoint与视图函数的映射关系存储到Map对象中。
add_url_rule方法里：self.view_functions[endpoint] = view_func添加endpoint和view_func的映射。

请求匹配过程：

请求进来时有WSGI服务器处理并调用了Flask app的__call__方法，再调用了wsgi_app方法；
wsgi_app方法里创建一个上下文对象：ctx = self.request_context(environ)。然后实例化了MapAdapter对象作为上下文对象属性；
wsgi_app方法里调用上下文对象的push方法：ctx.push()。这个方法主要使用MapAdapter对象的match方法。
MapAdapter对象的match方法，去调用rule对象的match方法。这个方法根据Map对象存储的路由信息去匹配当前请求的路由，得到了rule对象和参数对象放到上下文对象里。
wsgi_app方法里调用full_dispatch_request方法，然后里面再调用dispatch_request()方法；
dispatch_request()方法里：获取请求上下文对象，拿到里面的Rule对象，根据rule对象的endpoint属性从self.view_functions属性中获取对应的视图函数，然后把上下文中的参数传到视图函数中并调用视图函数处理请求，返回处理结果。

整个流程图如下：

在这里插入图片描述

其中上半部分项目启动时如何使用

Rule

和

Map

对象建立路由规则；下半部分是请求进来时，如何使用路由规则进行匹配。

标签： flask python 后端

本文转载自: https://blog.csdn.net/qq_43745578/article/details/129336596
版权归原作者 一根头发学一年 所有，如有侵权，请联系我们删除。

Flask源码篇：Flask路由规则与请求匹配过程

目录

1 启动时路由相关操作

（1）分析app.route()

（2）分析add_url_rule()

（3）分析Rule类

（4）分析Map类

（5）分析MapAdapter类

（6）分析 url_rule_class()

（7）分析map.add(rule)

2 请求进来时路由匹配过程

（1）分析wsgi_app

（2）分析request_context

（3）分析ctx.push

（4）分析full_dispatch_request

3 总结

发表评论

“Flask源码篇：Flask路由规则与请求匹配过程”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航