likes
comments
collection
share

flask源码分析(二)路由

作者站长头像
站长
· 阅读数 4

前言

本系列将对python几个流行的web框架源码进行分析。主要分析各个框架的特点、特色功能实现原理、以及一个请求在框架中的经过哪些处理。 文章不打算精读每一行代码,而是挑出框架比较有特色的地方深入研究,对于其他内容,采取粗读的形式。 预计会选取flask,fastapi,Django这三个框架。 每个框架各有特色,所以文章没有固定格式和必要篇目。

第一印象

先来看看flask中如何进行路由和视图函数绑定:

from flask import Flask

app = Flask(__name__)


@app.route('/route', methods=['GET'])
def route():
    return 'route'


@app.get('/get')
def get():
    return 'get'


@app.post('/post')
def post():
    return 'post'


def add_func():
    return 'add'


app.add_url_rule('/add', view_func=add_func)

if __name__ == '__main__':
    app.run(host='127.0.0.1')

上面展示了flask中常见的路由绑定方式。 通过@app.route装饰器,传入路由地址,请求方式(默认GET请求)。 通过查看源码可以知道,不管使用app.get,app.post还是app.route,最后都是调用了app.add_url_rule进行绑定:

# .../falsk/sansio/scaffold.py

# get和post是通过传入具体请求方法,给外部提供了方便的调用方式
@setupmethod
def get(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
    return self._method_route("GET", rule, options)

@setupmethod
def post(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:

    return self._method_route("POST", rule, options)

# 调用了route方法
def _method_route(
    self,
    method: str,
    rule: str,
    options: dict,
) -> t.Callable[[T_route], T_route]:
    if "methods" in options:
        raise TypeError("Use the 'route' decorator to use the 'methods' argument.")

    return self.route(rule, methods=[method], **options)
    
@setupmethod
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
    def decorator(f: T_route) -> T_route:
        endpoint = options.pop("endpoint", None)
        self.add_url_rule(rule, endpoint, f, **options)
        return f
    return decorator

可以看到route函数对业务函数进行装饰。route接收一个route路由地址,以及其他的options键值对参数。判断options有无endpoint,这个将用来进行参数匹配。这里如果没有传入endpoint,这里就会设置成空,在后续处理中会默认为视图函数名。

flask/sansio/scaffold.py提供了add_url_rule的函数签名,有两个实现,一个是提供上面普通调用方式,另一个是提供给蓝图。蓝图后续会继续分析。

@setupmethod
def add_url_rule(
    self,
    rule: str,
    endpoint: str | None = None,
    view_func: ft.RouteCallable | None = None,
    provide_automatic_options: bool | None = None,
    **options: t.Any,
) -> None:
    raise NotImplementedError

具体实现:

# .../flask/sansio/app.py
@setupmethod
def add_url_rule(
    self,
    rule: str,
    endpoint: str | None = None,
    view_func: ft.RouteCallable | None = None,
    provide_automatic_options: bool | None = None,
    **options: t.Any,
) -> None:
    if endpoint is None:
        endpoint = _endpoint_from_view_func(view_func)  # type: ignore
    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, 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 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  # type: ignore

    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"
                f" endpoint function: {endpoint}"
            )
        self.view_functions[endpoint] = view_func

通过上面的代码得知:

  • 如果没有设置请求方法methods,就默认是GET,并规定methods是包含字符串的列表。还会有一个required_method。如果传入的视图函数对象view_funcrequired_methods,就会通过methods |= required_methods合并请求方法。
  • 如果没有传入endpoint,就会调用_endpoint_from_view_func,返回视图函数对象的名称:
def _endpoint_from_view_func(view_func: t.Callable) -> str:
    """Internal helper that returns the default endpoint for a given
    function.  This always is the function name.
    """
    assert view_func is not None, "expected view func if endpoint is not provided."
    return view_func.__name__
  • 通过provide_automatic_options自动填充OPTIONS

接着是重点的路由绑定:

  • rule是self.url_rule_class返回的一个Rule对象,并将Rule对象存到url_map中,url_map是一个Map对象。 RuleMap后续会展开分析。

  • 这里会将视图函数通过endpoint作为key储存在view_functions这个字典中。这边会有一个endpoint是否绑定多个不同的函数。 这是什么意思呢? 首先看一下如果用route装饰同名的视图函数:

      @app.post('/post')
      def post():
          return 'post'
    
    
      @app.post('/post2')
      def post():
          return post
    

    这是会报错的,因为两个视图函数都是【post】,但是他们并不是同一个对象,然后源码中是通过if old_func is not None and old_func != view_func:进行判断,所以这个过程将会是"/post"的enpoint是【post】,第二次绑定的时候,"/post2"对应的enpoint也是【post】,那么,这将会开始判断两个视图函数是否是同一个对象,不是,就报错。 因此,假如需要在不同的路由中使用同一个函数,可以通过这种方式:

    def post():
        return 'post'
    
    app.add_url_rule('/post', view_func=post)
    app.add_url_rule('/post2', view_func=post)
    

    这时候,视图函数名字一样,并且是属于同一个对象,这样就不会抛出错误了。

    通过设置不同的endpoint也是可以避免错误的:

    @app.post('/post',endpoint='post')
    def post():
        return 'post'
    
    
    @app.post('/post2',endpoint='post2')
    def post():
        return post
    

    看样子,endpoint和实际的业务,以及对接口调用者不相干,只负责视图函数的映射。但是从请求的角度来说,我们是通过路由找到的视图函数。 endpoint好像和路由有着什么关系,这个后续会对比RuleMapendpoint三者的关系和作用。

path匹配view function

我们直接看一下当访问某个url地址的时候,flask是如何调用到对应的试图函数的:

# flask/app.py

def wsgi_app(
    self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
    ctx = self.request_context(environ)
    error: BaseException | None = 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 "werkzeug.debug.preserve_context" in environ:
            environ["werkzeug.debug.preserve_context"](_cv_app.get())
            environ["werkzeug.debug.preserve_context"](_cv_request.get())

        if error is not None and self.should_ignore_error(error):
            error = None

        ctx.pop(error)

def full_dispatch_request(self) -> Response:
    """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._got_first_request = True

    try:
        request_started.send(self, _async_wrapper=self.ensure_sync)
        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)

def dispatch_request(self) -> ft.ResponseReturnValue:
    req = request_ctx.request
    if req.routing_exception is not None:
        self.raise_routing_exception(req)
    
    # 获取请求中的Rule类
    rule: Rule = req.url_rule  # type: ignore[assignment]
    # 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
    view_args: dict[str, t.Any] = req.view_args  # type: ignore[assignment]
    # 根据Rule类中的endpoint匹配对应的视图函数
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) 

根据wsgi的协议,每一个请求过来,都会调用满足wsgi协议的可调用对象,在flask中就是wsgi_app。 最终调用的是dispatch_request。 flask会从request_ctx中获取请求上下文。请求上下文这个概念后续再进行展开,现在只要知道关于一个请求的所有信息,都会被储存在这个地方,包括了url地址,以及下面需要用到的endpoint。 最后flask通过return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) 返回具体的视图函数调用结果。 这里可以看到,通过Rule类中的endpoint匹配上文中提到的储存好的视图函数。

这一步得知,url地址通过Rule类匹配endpointendpoint匹配视图函数。这是在falsk中完成的。

Rule类

Rule的构建需要werkzueg完成。 上文中的add_url_rule函数中会初始化一个Rule类的实例:

# sansio/app.py
def add_url_rule(
    self,
    rule: str,
    endpoint: str | None = None,
    view_func: ft.RouteCallable | None = None,
    provide_automatic_options: bool | None = None,
    **options: t.Any,
) -> None:
       
    ........
    rule_obj = self.url_rule_class(rule, methods=methods, **options)
    rule_obj.provide_automatic_options = provide_automatic_options  # type: ignore[attr-defined]

    self.url_map.add(rule_obj)
    .......

它需要ruleurl path和endpoint,这也是后来可以根据path找到endpoint的原因。 重点在于self.url_map.add(rule_obj),这是Map类和Rule的关联。

Map类

和Rule类一样,Map类由werkzeug提供,flask会初始化一个Map类:

# sansio/app.py
self.url_map = self.url_map_class(host_matching=host_matching)
# 它会在add_url_rule中调用add函数

rule_obj = self.url_rule_class(rule, methods=methods, **options)
rule_obj.provide_automatic_options = provide_automatic_options  # type: ignore[attr-defined]

self.url_map.add(rule_obj)
# werkzeug/routing/map.py
def add(self, rulefactory: RuleFactory) -> None:
    for rule in rulefactory.get_rules(self):
        rule.bind(self)
        if not rule.build_only:
            self._matcher.add(rule)
        self._rules_by_endpoint.setdefault(rule.endpoint, []).append(rule)
    self._remap = True

# werkzeug/routing/rules.py
def bind(self, map: Map, rebind: bool = False) -> None:
    if self.map is not None and not rebind:
        raise RuntimeError(f"url rule {self!r} already bound to map {self.map!r}")
    self.map = map
    if self.strict_slashes is None:
        self.strict_slashes = map.strict_slashes
    if self.merge_slashes is None:
        self.merge_slashes = map.merge_slashes
    if self.subdomain is None:
        self.subdomain = map.default_subdomain
    self.compile()

可以看到,初始化一个Rule实例之后,会调用add函数进行绑定。 这里有点复杂,先看Rule.bind。它会调用compile进行编译。 详细的过程这里不展开,包括后续werkzeug如果解析url path,都不是本节的主要内容,后续会针对url解析的内容进行分析。 经过compile函数之后,Rule实例的_parts获得了赋值,这是一个列表,列表的元素是RulePart类的实例。

接下来是Map._matcher.add_matcherStateMachineMatcher的实例。调用add时会添加Rule类:

# werkzeug/routing/matcher.py

@dataclass
class State:
    dynamic: list[tuple[RulePart, State]] = field(default_factory=list)
    rules: list[Rule] = field(default_factory=list)
    static: dict[str, State] = field(default_factory=dict)

class StateMachineMatcher:
    def __init__(self, merge_slashes: bool) -> None:
        self._root = State()
        self.merge_slashes = merge_slashes
def add(self, rule: Rule) -> None:
    state = self._root
    for part in rule._parts:
        if part.static:
            state.static.setdefault(part.content, State())
            state = state.static[part.content]
        else:
            for test_part, new_state in state.dynamic:
                if test_part == part:
                    state = new_state
                    break
            else:
                new_state = State()
                state.dynamic.append((part, new_state))
                state = new_state
    state.rules.append(rule)

这就是当使用@app.route时候的全部过程了。但是现在还不是很明确,RuleMap的关系,要结合一下请求进来的时候发生了什么。

endpoint、Rule、Map

上文提到,当请求进来的时候,通过wsgi_app,最后是调用了self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) 这是通过Rule获得了endpoint,用endpoint去匹配view_functions的视图函数。 那么在此之前,还需要完成如果从url path到Rule这一步。

可以在代码中看到Rule对象都是从请求上下文中获取到的:

req = request_ctx.request
rule: Rule = req.url_rule 

请求上下文现在不展开分析,只要知道,它是一个栈,每次请求过来,都会将所需要的内容储存在一个对象中,并压入栈。这里所需的内容,就包括了url path。

# flask/app.py
def wsgi_app(
    self, environ: WSGIEnvironment, start_response: StartResponse
) -> cabc.Iterable[bytes]:
    ctx = self.request_context(environ)
    error: BaseException | None = 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)
       ......

看一下push中发生了什么:

# flask/ctx.py
def push(self) -> None:
    ....
    if self.url_adapter is not None:
        self.match_request()
        
def match_request(self) -> None:
    try:
        result = self.url_adapter.match(return_rule=True)  # type: ignore
        self.request.url_rule, self.request.view_args = result  # type: ignore
    except HTTPException as e:
        self.request.routing_exception = e

可以看到这一步,就将Rule类以及请求view_args绑定在request上面,然后就是上面的获取到。那么关键在url_adapter.match这边。

# flask/ctx.py
class RequestContext:
    def __init__(
        self,
        app: Flask,
        environ: WSGIEnvironment,
        request: Request | None = None,
        session: SessionMixin | None = None,
    ) -> None:
        self.url_adapter = None
        try:
            self.url_adapter = app.create_url_adapter(self.request)
        except HTTPException as e:
            self.request.routing_exception = e
        ....

# flask/app.py
def create_url_adapter(self, request: Request | None) -> MapAdapter | None:
    .......
    if self.config["SERVER_NAME"] is not None:
        return self.url_map.bind(
            self.config["SERVER_NAME"],
            script_name=self.config["APPLICATION_ROOT"],
            url_scheme=self.config["PREFERRED_URL_SCHEME"],
        )

    return None


# werkzeug/routing/map.py
def bind_to_environ(
    self,
    environ: WSGIEnvironment | Request,
    server_name: str | None = None,
    subdomain: str | None = None,
) -> MapAdapter:
    ....

    def _get_wsgi_string(name: str) -> str | None:
        val = env.get(name)
        if val is not None:
            return _wsgi_decoding_dance(val)
        return None

    script_name = _get_wsgi_string("SCRIPT_NAME")
    path_info = _get_wsgi_string("PATH_INFO")
    query_args = _get_wsgi_string("QUERY_STRING")
    return Map.bind(
        self,
        server_name,
        script_name,
        subdomain,
        scheme,
        env["REQUEST_METHOD"],
        path_info,
        query_args=query_args,
    )

def bind(
    self,
    server_name: str,
    script_name: str | None = None,
    subdomain: str | None = None,
    url_scheme: str = "http",
    default_method: str = "GET",
    path_info: str | None = None,
    query_args: t.Mapping[str, t.Any] | str | None = None,
) -> MapAdapter:
    ......
    return MapAdapter(
        self,
        f"{server_name}{port_sep}{port}",
        script_name,
        subdomain,
        url_scheme,
        path_info,
        default_method,
        query_args,
    )

可以看到,最后返回的是一个MapAdapter对象。它的match函数最后会调用StateMachineMatchermatch函数,返回Rule对象和view_args

def match(
    self, domain: str, path: str, method: str, websocket: bool
) -> tuple[Rule, t.MutableMapping[str, t.Any]]:
    # To match to a rule we need to start at the root state and
    # try to follow the transitions until we find a match, or find
    # there is no transition to follow.

    have_match_for = set()
    websocket_mismatch = False

    def _match(
        state: State, parts: list[str], values: list[str]
    ) -> tuple[Rule, list[str]] | None:
        .......
        if parts == [""]:
            for rule in state.rules:
                if rule.strict_slashes:
                    continue
                if rule.methods is not None and method not in rule.methods:
                    have_match_for.update(rule.methods)
                elif rule.websocket != websocket:
                    websocket_mismatch = True
                else:
                    return rule, values

        return None

所以,endpointRuleMap三者关系可以这么理解:

  • 程序启动的时候,通过add_url_rule做了这几件事:
    • 初始化Rule对象,这里Rule实例保存了endpoint
    • Rule进行compile,存入StateMachineMatcherStateMachineMatcher存在Map
  • 请求过来的时候:
    • 通过请求上下文获取到请求信息,包括url path
    • Map根据url path找到Rule
    • 通过Rule对象的endpoint去匹配对应的视图函数

所以三者的关系是:通过Map找到Rule,通过Rule找到endpoint,通过endpoint找到定义好的视图函数。

总结

本篇主要进行了flask的路由分析,从最开始的@app.route到最后请求的过程。其中涉及到的其他的内容,比如上下文,具体如何解析url和请求参数,并不详细展开。上下文的内容会在下一篇目开始分析。flask路由系统的其他细节,包括解析url,匹配Rule,正则转换器等等会在系列最后进行补充。

flask的路由复杂的地方在于Rule类和Map类的关系。从分析结果来看,endpointRule的匹配是由flask完成的,而RuleMap类的匹配是由werkzeug提供的。

转载自:https://juejin.cn/post/7335466697396912140
评论
请登录