深入理解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()
