• 周四. 12月 1st, 2022

5G编程聚合网

5G时代下一个聚合的编程学习网

热门标签

Flask之路由系统

admin

11月 28, 2021

7.路由系统

@app.route():将url地址绑定到视图函数的装饰器

参数:

methods: 当前url允许访问的请求方式

endpoint: 反向url地址,默认为视图函数名(url_for)

defaults : 视图函数的参数默认值{“nid”:1}

strict_slashes : url地址结尾符”/”的控制 False : 无论结尾 “/” 是否存在均可以访问 , True : 结尾必须不能是 “/”

redirect_to : url地址重定向

subdomain : 子域名前缀 subdomian=”DragonFire” 这样写可以得到 DragonFire.oldboyedu.com 前提是app.config[“SERVER_NAME”] = “oldboyedu.com”

[email protected]()装饰器的参数

methods: 当前url地址允许访问的请求方式

@app.route("/info", methods=["GET", "POST"])
def stu_info():
	stu_id = int(reuqest.agrs['id'])
	return "OK"

endpoint: 反向url地址,默认为视图函数名(url_for)

from flask import url_for


@app.route("/info", methods=["GET", "POST"], endpoint="r_info")
def student_info():
    print(url_for("r_info"))  # /info
    stu_id = int(request.args["id"])
    return "OK"

defaults : 视图函数的参数默认值{“nid”:1}

from flask import url_for

@app.route("/info", methods=["GET", "POST"], endpoint="r_info", defaults={"nid": 100})
def student_info(nid):
    print(url_for("r_info"))  # /info
    print(nid)  # 100
    return "OK"

strict_slashes : url地址结尾符”/”的控制 False : 无论结尾 “/” 是否存在均可以访问 , True : 结尾必须不能是 “/”

# 访问地址 : /info 
@app.route("/info", strict_slashes=True)
def student_info():
    return "Hello info"


# 访问地址 : /infos  or  /infos/
@app.route("/infos", strict_slashes=False)
def student_infos():
    return "Hello infos"

redirect_to : url地址重定向

# 访问地址 : /info 浏览器跳转至 /infos
@app.route("/info", strict_slashes=True, redirect_to="/infos")
def student_info():
    return "OK"

@app.route("/infos", strict_slashes=False)
def student_infos():
    return "OK"

subdomain : 子域名前缀 subdomian=”Seven” 这样写可以得到 Seven.duke.com 前提是app.config[“SERVER_NAME”] = “duke.com”

app.config["SERVER_NAME"] = "duke.com"

@app.route("/info",subdomain="Seven")
def student_info():
    return "OK"

# 访问地址为:  Seven.duke.com/info

2.动态参数路由

from flask import url_for


# 访问地址 : http://127.0.0.1:5000/info/1
@app.route("/info/<int:nid>", methods=["GET", "POST"], endpoint="r_info")
def student_info(nid):
    print(url_for("r_info",nid=2))  # /info/2
    return f"访问的infoid为:{nid}"  # Python3.6的新特性 f"{变量名}"

PS:在url_for的时候,一定要将动态参数名+参数值添加进去,否则会抛出参数错误的异常

3.一个路由的问题

两个模块

main

news

需要将news中的内容加载到main中,而news中路由基于main中生成的app,会产生的问题:循环导入的问题题。

运行main.py报错, ImportError: cannot import name ‘news’, 就是由于循环导入引起的。

# main.py
from flask import Flask
from news import news

app = Flask(__name__)

@app.route("/")
def index():
    return "index"

@app.route("/users")
def users():
    return "users"

if __name__ == "__main__":
    app.run(debug=True)
# news.py
from main import app
    @app.route("/news")
    def news():
        return "news"

产生了一个问题:怎么取进行分模块开发,路由的设置怎么处理?

解决办法:app作为共有数据,每一个模块都引入app进行路由的派发,这样就不会引起循环引用的问题了。又来:路由派发不会出问题,但是对应路由的视图函数每个模块没有定义,使用urf_for(view_name)反向生成路由时几个模块中的视图名重名怎么办?

为了解决上面几个问题,引出了蓝图

4.蓝图

一个xxx网站,可能用到首页模块、用户模块、后台模块等等

在项目开发过程中,需要把项目根据相关的功能划分为对应的模块,通过模块的划分可以更好的组织项目的目录结构,使项目的整个框架更加清晰

目录形式的蓝图:

# User模块蓝图目录
-- users
	- __init__.py
	- views.py

# __init__.py
from flask import Blueprint
users_blu = BluePrint("users", __name__)

from . import views

# views.py
from . import users_blu
# 使用蓝图注册路由
@users_blu.route("/users")
def users():
    return "users"

# main.py
from flask import Flask
from users import users_blue

app = Flask(__name__)
app.register_blueprint(users_blue)

# 在前端的url_for方向生成users视图函数映射的路由
{{url_for('users_blue.users')}}

5.蓝图对象的参数

可以有模块自己跌静态文件存储目录,模板文件的存储目录。

users_blu = Blueprint(

name,

import_name,

static_folder=“static”, # 蓝图中静态文件存储目录

static_url_path=”/users/static”, # 访问蓝图中静态文件url地址前缀

template_folder=“templates”, # 蓝图中模板文件的存储目录

# url_prefix=”/users” # 统一该模块下资源请求的前缀
)

6.flask路由源码剖析

转载: http://cizixs.com/2017/01/12/flask-insight-routing

路由:根据请求的URL找到对应的处理函数,也就是URL->view_function(视图函数)的映射。字典可以满足需求,对于静态路由至于要通过路由这个key值找到对应的视图函数这个value值,找不到value值就报404错误。而对于动态路由,更复杂的匹配逻辑,flask中的路由系统是怎么构建的?

在分析路由匹配过程之前,我们先来看看 flask 中,构建这个路由规则的两种方法:

  1. 通过 [@app.route](mailto:@app.route)()` decorator,比如文章开头给出的 hello world 例子

  2. 通过

app.add_url_rule

,这个方法的签名为

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

,参数的含义如下:

  • rule: url 规则字符串,可以是静态的 /path,也可以包含 /
  • endpoint:要注册规则的 endpoint,默认是 view_func 的名字
  • view_func:对应 url 的处理函数,也被称为视图函数
# 两种方法等价,如下所示:

@app.route('/')
def hello():
    return "hello"

# 等价
def hello():
    return "hello"
app.add_url_rule('/', 'hello', hello)

1.route装饰器

调用了app对象的route方法来装饰hello视图函数

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
"""
1.在用app.route装饰hello_world视图函数的时候,实际上app.route中还可以添加一些参数。
2.比如指定请求的方法的变量:methods=["GET","POST"]以及指定视图函数的endpoint,相当于Django中视图函数的别名等
3.rule参数相当于hello_world视图函数中的"/"路径,options参数中包含methods和endpoint等
4.在route装饰器里,返回decorator闭包函数。
5.在decorator闭包函数中,先从options中获取endpoint的值,endpoint的值默认为None
6.然后调用self.add_url_rule内部方法处理传递的参数rule,endpoint,f等,在这里self指的是app这个对象
"""

2.add_url_rule方法

  1. 视图函数中没有指定endpoint时,程序会调用_endpoint_from_view_func方法为endpoint赋值 ,而 _endpoint_from_view_func实际上返回的就是view_func函数的函数名。 所以此时,在options这个字典中有一个键为endpoint,对应的值为view_func函数的函数名
def _endpoint_from_view_func(view_func):
   assert view_func is not None, 'expected view func if endpoint ' 
                                 'is not provided.'
   return view_func.__name__
  1. 接着,程序从函数中获取”required_methods”的值,并进行去重,默认得到一个空集合

再对methods和required_methods进行"|="操作,也就是按位或运算。 对methods和required_methods进行按位或运算,实际上就是把required_methods的值添加到methods方法集合里

  1. 接着程序调用self.url_rule_class方法处理rule(也就是”/”),methods和options字典

得到rule这个对象,在这里self同样指的是app这个对象

可以看到,url_rule_class指向的是Rule这个类的内存地址

url_rule_class = Rule

  1. 然后用Map类实例化得到self.url_map对象,调用self.url_map对象中的add方法处理rule这个对象self.url_map = Map()

总结:

可以看到它主要做的事情就是更新 self.url_mapself.view_functions 两个变量。找到变量的定义,发现 url_mapwerkzeug.routeing:Map 类的对象,rulewerkzeug.routing:Rule 类的对象,view_functions 就是一个字典。这和我们之前预想的并不一样,这里增加了 RuleMap 的封装,还把 urlview_func 保存到了不同的地方。

需要注意的是:每个视图函数的 endpoint 必须是不同的,否则会报 AssertionError

@setupmethod
    def add_url_rule(self, rule, endpoint=None, view_func=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.
        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

3.werkzeug 路由逻辑

事实上,flask 核心的路由逻辑是在 werkzeug 中实现的。所以在继续分析之前,我们先看一下 werkzeug 提供的路由功能

>>> m = Map([
...     Rule('/', endpoint='index'),
...     Rule('/downloads/', endpoint='downloads/index'),
...     Rule('/downloads/<int:id>', endpoint='downloads/show')
... ])
>>> urls = m.bind("example.com", "/")
>>> urls.match("/", "GET")
('index', {})
>>> urls.match("/downloads/42")
('downloads/show', {'id': 42})

>>> urls.match("/downloads")
Traceback (most recent call last):
  ...
RequestRedirect: http://example.com/downloads/
>>> urls.match("/missing")
Traceback (most recent call last):
  ...
NotFound: 404 Not Found

上面的代码演示了 werkzeug 最核心的路由功能:添加路由规则(也可以使用 m.add),把路由表绑定到特定的环境(m.bind),匹配url(urls.match)。正常情况下返回对应的 endpoint 名字和参数字典,可能报重定向或者 404 异常。

可以发现,endpoint 在路由过程中非常重要werkzeug 的路由过程,其实是 url 到 endpoint 的转换:通过 url 找到处理该 url 的 endpoint。至于 endpoint 和 view function 之间的匹配关系,werkzeug 是不管的,而上面也看到 flask 是把这个存放到字典中的。

4.flask路由实现

回头看 dispatch_request,继续探寻路由匹配的逻辑:

这个方法做的事情就是找到请求对象 request,获取它的 endpoint,然后从 view_functions 找到对应 endpointview_func ,把请求参数传递过去,进行处理并返回。view_functions 中的内容,我们已经看到,是在构建路由规则的时候保存进去的;那请求中 req.url_rule 是什么保存进去的呢?它的格式又是什么?

我们可以先这样理解:_request_ctx_stack.top.request 保存着当前请求的信息,在每次请求过来的时候,flask 会把当前请求的信息保存进去,这样我们就能在整个请求处理过程中使用它。至于怎么做到并发情况下信息不会相互干扰错乱,可以看转载的其他文章。

def dispatch_request(self):
    """Does the request dispatching.  Matches the URL and returns the
    return value of the view or error handler.  This does not have to
    be a response object.  In order to convert the return value to a
    proper response object, call :func:`make_response`.
    """

    req = _request_ctx_stack.top.request
    if req.routing_exception is not None:
        self.raise_routing_exception(req)
    rule = req.url_rule

    # dispatch to the handler for that endpoint
    return self.view_functions[rule.endpoint](**req.view_args)

_request_ctx_stack 中保存的是 RequestContext 对象,它出现在 flask/ctx.py 文件中,和路由相关的逻辑如下:

在初始化的时候,会调用 app.create_url_adapter 方法,把 appurl_map 绑定到 WSGI environ 变量上(bind_to_environ 和之前的 bind 方法作用相同)。最后会调用 match_request 方法,这个方式调用了 url_adapter.match 方法,进行实际的匹配工作,返回匹配的 url rule。而我们之前使用的 url_rule.endpoint 就是匹配的 endpoint 值。

整个 flask 的路由过程就结束了,总结一下大致的流程:

  • 通过 [@app.route](mailto:@app.route)或者app.add_url_rule` 注册应用 url 对应的处理函数
  • 每次请求过来的时候,会事先调用路由匹配的逻辑,把路由结果保存起来
  • dispatch_request 根据保存的路由结果,调用对应的视图函数
class RequestContext(object):
    def __init__(self, app, environ, request=None):
        self.app = app
        self.request = request
        self.url_adapter = app.create_url_adapter(self.request)
        self.match_request()

    def match_request(self):
        """Can be overridden by a subclass to hook into the matching
        of the request.
        """
        try:
            url_rule, self.request.view_args = 
                self.url_adapter.match(return_rule=True)
            self.request.url_rule = url_rule
        except HTTPException as e:
            self.request.routing_exception = e


class Flask(_PackageBoundObject):
    def create_url_adapter(self, request):
        """Creates a URL adapter for the given request.  The URL adapter
        is created at a point where the request context is not yet set up
        so the request is passed explicitly.
        """
        if request is not None:
            return self.url_map.bind_to_environ(request.environ,
                server_name=self.config['SERVER_NAME'])

5.match实现

匹配的逻辑:

用实现 compile 的正则表达式去匹配给出的真实路径信息,把所有的匹配组件转换成对应的值,保存在字典中(这就是传递给视图函数的参数列表)并返回。

讲完了 flask 的路由流程,但是还没有讲到最核心的问题:werkzeug 中是怎么实现 match 方法的。Map 保存了 Rule 列表,match 的时候会依次调用其中的 rule.match 方法,如果匹配就找到了 match。Rule.match 方法的代码如下:

def match(self, path):
        """Check if the rule matches a given path. Path is a string in the
        form ``"subdomain|/path(method)"`` and is assembled by the map.  If
        the map is doing host matching the subdomain part will be the host
        instead.

        If the rule matches a dict with the converted values is returned,
        otherwise the return value is `None`.
        """
        if not self.build_only:
            m = self._regex.search(path)
            if m is not None:
                groups = m.groupdict()

                result = {}
                for name, value in iteritems(groups):
                    try:
                        value = self._converters[name].to_python(value)
                    except ValidationError:
                        return
                    result[str(name)] = value
                if self.defaults:
                    result.update(self.defaults)

                return result
# groupdict()
# 它返回一个字典,包含所有经命名的匹配子群,键值是子群名。

m = re.match(r'(?P<user>w+)@(?P<website>w+).(?P<extension>w+)','[email protected]')
m.groupdict() # {'website': 'hackerrank', 'user': 'myname', 'extension': 'com'}

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注