深入理解Flask路由 (3) - 动态 url 及转换器

本篇讲解动态 url 和转换器的用法及原理。

动态 url 实现原理

动态 url 由 werkzeug 通过转换器 (converter) 来实现，为说明动态 url 的使用方法，我们先从简单的示例开始，逐步展开。假设我们要编写一个向手机号码发送短信的 Flask 程序，有如下一段代码：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return 'Index page'

@app.route('/message/<mobile_number>')
def send_message(mobile_number):
    return 'Message was sent to {}'.format(mobile_number)

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

由于 Flask 对路由过程作了较抽象的封装，并不容易看出完整的过程，为了便于理解，我们直接用下面的一段代码来说明在 werkzeurg 中路由绑定 (bind) 和匹配 (match) 的过程。

from werkzeug.routing import Rule, Map
from werkzeug.serving import run_simple
from werkzeug.exceptions import HTTPException

rules = [
    Rule('/', endpoint='index'),
    Rule('/message/<mobile_number>', endpoint='mobile')
]
url_map = Map(rules)

def application(environ, start_response):
    urls = url_map.bind_to_environ(environ)
    try:
        endpoint, args = urls.match()
    except HTTPException as ex:
        return ex(environ, start_response)

    headers = [('Content-Type', 'text/plain')]
    start_response('200 OK', headers)

    body = 'Rule points to {} with arguments {}' \
        .format(endpoint, args).encode('utf-8')
    return [body]

if __name__ == "__main__":
    run_simple('localhost', 5000, application)

这段代码能架起一个简单的 Web 服务器。当客户端从 /messages/mobile_number 发起 GET 请求，程序返回如下信息：

Rule points to mobile with arguments {'mobile_number': '13811112222'}

这段代码展示了 werkzueg 路由过程的三大阶段：

创建 Rule 和 Map 的实例上篇已经讲过。本篇从第二步开始讲解。

绑定到特定环境

bind_to_environ() 方法在内部调用了 Map.bind() 方法， Map.bind() 方法创建 MapAdapter 的实例。MapAdapter 类负责 URL 匹配的工作。

以下是关键代码及说明。

• bind_to_environ() 方法在内部调用了 Map.bind() 方法：

def bind_to_environ(self, environ, server_name=None, subdomain=None):   
    # 其他代码略
    
    return Map.bind(
        self,
        server_name,
        script_name,
        subdomain,
        environ["wsgi.url_scheme"],
        environ["REQUEST_METHOD"],
        path_info,
        query_args=query_args,
    )

• Map.bind() 方法创建 MapAdapter的实例：

def bind(
    self,
    server_name,
    script_name=None,
    subdomain=None,
    url_scheme="http",
    default_method="GET",
    path_info=None,
    query_args=None,
):
   # 其他代码略
   
    return MapAdapter(
        self,
        server_name,
        script_name,
        subdomain,
        url_scheme,
        path_info,
        default_method,
        query_args,
    )

MapAdapter.match() 方法

该方法的作用是，传入 path_info 和 method，返回 tuple 类型包括 endpoint + arguments 的信息或者 rule + arguments 信息 (You get a tuple in the form (endpoint, arguments) if there is a match (unless return_rule is True, in which case you get a tuple in the form (rule, arguments)))。该方法内部调用 Rule.match() 方法：

def match(self, path_info=None, method=None, return_rule=False, query_args=None):
    
    self.map.update()
    if path_info is None:
        path_info = self.path_info
    else:
        path_info = to_unicode(path_info, self.map.charset)
    if query_args is None:
        query_args = self.query_args
    method = (method or self.default_method).upper()

    path = u"%s|%s" % (
        self.map.host_matching and self.server_name or self.subdomain,
        path_info and "/%s" % path_info.lstrip("/"),
    )

    have_match_for = set()
    for rule in self.map._rules:
        try:
            #-----------------------------------
            # 内部调用 Rule.match()方法
            #-----------------------------------
            rv = rule.match(path, method)
        except RequestSlash:
            raise RequestRedirect(
                self.make_redirect_url(
                    url_quote(path_info, self.map.charset, safe="/:|+") + "/",
                    query_args,
                )
            )
        except RequestAliasRedirect as e:
            raise RequestRedirect(
                self.make_alias_redirect_url(
                    path, rule.endpoint, e.matched_values, method, query_args
                )
            )
        if rv is None:
            continue
        if rule.methods is not None and method not in rule.methods:
            have_match_for.update(rule.methods)
            continue

        if self.map.redirect_defaults:
            redirect_url = self.get_default_redirect(rule, method, rv, query_args)
            if redirect_url is not None:
                raise RequestRedirect(redirect_url)

        if rule.redirect_to is not None:
            if isinstance(rule.redirect_to, string_types):

                def _handle_match(match):
                    value = rv[match.group(1)]
                    return rule._converters[match.group(1)].to_url(value)

                redirect_url = _simple_rule_re.sub(_handle_match, rule.redirect_to)
            else:
                redirect_url = rule.redirect_to(self, **rv)
            raise RequestRedirect(
                str(
                    url_join(
                        "%s://%s%s%s"
                        % (
                            self.url_scheme or "http",
                            self.subdomain + "." if self.subdomain else "",
                            self.server_name,
                            self.script_name,
                        ),
                        redirect_url,
                    )
                )
            )
        # 如果设置 return_rule，返回 rule+arguments(tuple)
        if return_rule:
            return rule, rv
        # 否则返回 endpoint+arguments(tuple)
        else:
            return rule.endpoint, rv

    if have_match_for:
        raise MethodNotAllowed(valid_methods=list(have_match_for))
    raise NotFound()

比如在本例中, 请求的 path 为 /message/13811112222，调用 match() 方法后，返回 {'mobile_number' : '13811112222'} (dict 类型)

转换器与动态 url

Flask 中实现动态 url 的方法是通过 Map 类的转换器 (converter）来实现，converter 中定义了 regular expression，在创建 Map 实例的时候，添加 Rule 的时候会有一系列方法调用：

• Map 初始化方法调用 Map.add() 方法
• Map.add() 方法针对每一个 Rule， 调用 Rule.bind() 方法
• Rule.bind() 方法调用 Rule.compile() 方法
• 在 Rule.compile() 方法中，定义了一个内部方法 _build_regex() 将 URL 解析为 converter, arguments 和 variable:

然后根据 Rule 中 converter 名称调用对应的 converter，没有指定 converter 名称则默认为 UnicodeConverter:

Rule 的格式是<converter(arguments):name> ，符合规则的格式才能被正确解析，werkzueg 实现了 7 中预定义的 converter，能满足绝大部分需求。在 routing.py 中，我们可以看到如下的代码：

DEFAULT_CONVERTERS = {
    "default": UnicodeConverter,
    "string": UnicodeConverter,
    "any": AnyConverter,
    "path": PathConverter,
    "int": IntegerConverter,
    "float": FloatConverter,
    "uuid": UUIDConverter,
}

class Map(object):
    default_converters = ImmutableDict(DEFAULT_CONVERTERS)
    # ...

7 种 converter，都直接或者间接继承自 BaseConverter。BaseConverter 类的代码如下：

class BaseConverter(object):
    """Base class for all converters."""

    regex = "[^/]+"
    weight = 100

    def __init__(self, map):
        self.map = map

    def to_python(self, value):
        return value

    def to_url(self, value):
        if isinstance(value, (bytes, bytearray)):
            return _fast_url_quote(value)
        return _fast_url_quote(text_type(value).encode(self.map.charset))

每一种 converter 的 __init__() 方法确定了 converter 可以使用哪些 arguments。比如 UnicodeConverter 的 __init__() 方法是这样的：

def __init__(self, map, minlength=1, maxlength=None, length=None):
    # 代码略

所以我们在使用 UnicodeConverter 的时候能够使用 length 、minlenght 和 maxlenght 这些参数。比如下面的示例：

rules = [
    Rule('/message/<string(length=11):mobile_number>', endpoint='mobile')
]

BaseConveter 类的 to_python() 方法在 MapAdapter.match() 方法中匹配成功后被调用，将请求的 path 中动态 url 部分解析出 argument value， 传递给该方法的value 参数。在上面的示例中，13811112222 手机号码被解析出来，传递给 to_python() 方法。

to_url() 方法在 url_for() 函数反向构建url 的时候，将 arguments 参数传递给该方法。后面结合具体的示例来帮助大家理解。

UnicodeConverter 是默认的转换器，用于实现 string 类型的动态 url。

自定义转换器

刚刚给出的例子没有针对手机号码的校验规则，假设我们要对请求中传递的手机号码进行校验，可以用自定义转换器来实现。如果只是想增加校验规则，在转换器的 __init__() 方法中改写 BaseConverter 的 regex 属性。

自定义转换器代码如下：

# CustomConverter.py

from werkzeug.routing import BaseConverter

class MobileConverter(BaseConverter):
    def __init__(self, map):
        BaseConverter.__init__(self, map)
        self.regex = r'1[35678]\d{9}'

然后在创建 Map 的时候 converters 参数从自定义转换器赋值：

from CustomConverter import MobileConverter

mobile_converter = [{'mobile', MobileConverter}]
rules = [
    Rule('/', endpoint='index'),
    Rule('/message/<mobile:mobile_number>', endpoint='mobile')
]

url_map = Map(rules, converters=mobile_converter)

to_python() 方法如何使用呢？假设为了增加友好性，将返回到前端的手机号码用 - 分割，比如 13811112222 显示为 138-1111-2222。根据刚才的说明, to_python() 方法在匹配成功后将被调用，所以可以改写 BaseConverter 的 to_python() 方法，编写如下代码：

from werkzeug.routing import BaseConverter

class MobileConverter(BaseConverter):
    def __init__(self, map):
        BaseConverter.__init__(self, map)
        self.regex = r'1[35678]\d{9}'

    def to_python(self, value):
        return '{}-{}-{}'.format(value[:3], value[3:7], value[7:12])

to_url() 方法在 url_for() 函数中被调用，url_for() 函数的 argument 参数被传递给该方法的 value 参数。下面的示例演示了 to_url() 的用法。

from werkzeug.routing import BaseConverter

class MobileConverter(BaseConverter):
    def __init__(self, map):
        BaseConverter.__init__(self, map)
        self.regex = r'1[35678]\d{9}'

    def to_python(self, value):
        return '{}-{}-{}'.format(value[:3], value[3:7], value[7:12])

    def to_url(self, value):
        print(value)
        return value

from flask import Flask, url_for
from CustomConverter import MobileConverter

app = Flask(__name__)
app.url_map.converters['mobile'] = MobileConverter

@app.route('/')
def index():
    return 'Index page'

@app.route('/message/<mobile:mobile_number>')
def send_message(mobile_number):
    print(url_for('send_message', mobile_number='13833334444'))
    return 'Message was sent to {}'.format(mobile_number)

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