Manjusaka

Manjusaka

你所不知道的 Flask 第1部分:路由初探

前言#

我自己都記不清楚上一次寫博客是什麼時候了(笑),上一次挖的坑現在還沒填完,乾脆,開個新坑吧,你不知道的 Flask ,記錄下自己用 Flask 過程中一些很好玩的東西,當然很大可能我又會中途棄坑

開篇#

引子#

之前遇到一個很奇怪的需求,需要在 flask 中支持正則表達式比如,@app.route('/api/(.*?)') 這樣,在視圖函數被調用的時候,能傳入 URL 中正則匹配的值。不過 Flask 路由中默認不支持這樣的方法,那麼我們該怎麼辦?我們先思考五分鐘吧?

好了,我先給出解決方案吧

from flask import Flask
from werkzeug.routing import BaseConverter
class RegexConverter(BaseConverter):
    def __init__(self, map, *args):
        self.map = map
        self.regex = args[0]


app = Flask(__name__)
app.url_map.converters['regex'] = RegexConverter

在經過這樣的設置後我們便可以按照我們剛才的需求寫代碼了


@app.route('/docs/model_utils/<regex(".*"):url>')
def hello(url=None):

    print(url)

在這裡,我們函數中傳入的 url 變量,就是我們代碼中所匹配到的值

但是為什麼這樣就 OK 了呢?

詳解#

首先,我們要弄清楚一個東西,Flask 是 基於 Werkzurg 的一個框架,Flask 的 Route 機制基於 Werkzurg 上更進一步封裝所得到的,OK,我們上面所以實現的 Converter 便是利用了 Werkzurg 中的 Route 的特性

好了,我先給出官方文檔 custom-converters

然後我們來仔細講講,

首先,Werkzurg 中存在著一種機制叫做 Converter ,簡而言之就是通過一定的特殊語法,將 URL 中的特定部分,轉化成特定的 Python 變量,其語法格式為 /url/<converter_name("表達式"):變量名> 看起來有點複雜對吧,OK 用我們之前的例子來講一下吧,你看,我們之前定義了一個 '/docs/model_utils/<regex(".*"):url>' 的 URL ,其中後面部分就是利用了我們提到的 Converter 語法。具體的含義是,這個部分的 url 交給 regex 這個 Converter 來處理,最終生成的變量名為 url

好了,我們來說說自定義 Converter 參數中的注意事項,在構建一個自己的 Converter 過程中,我們將按照如下的方式編寫代碼

class RegexConverter(BaseConverter):
    def __init__(self, map, regex,*args):
        self.map = map
        self.regex = regex

map 是指 werkzurg.routing 中的 Map 對象,而 regex 則是指你所寫的表達式。其中 map 的作用我們將放在下一章進行講解,(又立 flag 了,笑)。

好了這裡差不多完成了,我們來看看 Flask 喔,不,werkzurg 中怎麼實現的這樣的方法吧

簡明代碼剖析#

最前面,你首先得有一點 flask 裝飾器路由的知識,詳情可以參考這篇文章,菜鳥閱讀 Flask 源碼系列(1):Flask 的 router 初探

首先在 werkzurg 框架的 routing 文件中,存在著這樣一段代碼


_rule_re = re.compile(r'''
    (?P<static>[^<]*)                           # static rule data
    <
    (?:
        (?P<converter>[a-zA-Z_][a-zA-Z0-9_]*)   # converter name
        (?:\((?P<args>.*?)\))?                  # converter arguments
        \:                                      # variable delimiter
    )?
    (?P<variable>[a-zA-Z_][a-zA-Z0-9_]*)        # variable name
    >
''', re.VERBOSE)
_simple_rule_re = re.compile(r'<([^>]+)>')
_converter_args_re = re.compile(r'''
    ((?P<name>\w+)\s*=\s*)?
    (?P<value>
        True|False|
        \d+.\d+|
        \d+.|
        \d+|
        \w+|
        [urUR]?(?P<stringval>"[^"]*?"|'[^']*')
    )\s*,
''', re.VERBOSE | re.UNICODE)

def parse_converter_args(argstr):
    argstr += ','
    args = []
    kwargs = {}

    for item in _converter_args_re.finditer(argstr):
        value = item.group('stringval')
        if value is None:
            value = item.group('value')
        value = _pythonize(value)
        if not item.group('name'):
            args.append(value)
        else:
            name = item.group('name')
            kwargs[name] = value

    return tuple(args), kwargs


def parse_rule(rule):
    """Parse a rule and return it as generator. Each iteration yields tuples
    in the form ``(converter, arguments, variable)``. If the converter is
    `None` it's a static url part, otherwise it's a dynamic one.

    :internal:
    """
    pos = 0
    end = len(rule)
    do_match = _rule_re.match
    used_names = set()
    while pos < end:
        m = do_match(rule, pos)
        if m is None:
            break
        data = m.groupdict()
        if data['static']:
            yield None, None, data['static']
        variable = data['variable']
        converter = data['converter'] or 'default'
        if variable in used_names:
            raise ValueError('variable name %r used twice.' % variable)
        used_names.add(variable)
        yield converter, data['args'] or None, variable
        pos = m.end()
    if pos < end:
        remaining = rule[pos:]
        if '>' in remaining or '<' in remaining:
            raise ValueError('malformed url rule: %r' % rule)
        yield None, None, remaining

首先,_rule_re 以及 _converter_args_re 兩段是很騷的正則表達式,不過作者已經給出了足夠的註釋,大家可以對照著正則表達式的語法進行學習一個,然後 parse_converter_args 以及 parse_rule 則是利用正則表達式對其進行解析操作。

OK,我們緊接著往下查看


    def compile(self):
        """Compiles the regular expression and stores it."""
        assert self.map is not None, 'rule not bound'

        if self.map.host_matching:
            domain_rule = self.host or ''
        else:
            domain_rule = self.subdomain or ''

        self._trace = []
        self._converters = {}
        self._weights = []
        regex_parts = []

        def _build_regex(rule):
            for converter, arguments, variable in parse_rule(rule):
                if converter is None:
                    regex_parts.append(re.escape(variable))
                    self._trace.append((False, variable))
                    for part in variable.split('/'):
                        if part:
                            self._weights.append((0, -len(part)))
                else:
                    if arguments:
                        c_args, c_kwargs = parse_converter_args(arguments)
                    else:
                        c_args = ()
                        c_kwargs = {}
                    convobj = self.get_converter(
                        variable, converter, c_args, c_kwargs)
                    regex_parts.append('(?P<%s>%s)' % (variable, convobj.regex))
                    self._converters[variable] = convobj
                    self._trace.append((True, variable))
                    self._weights.append((1, convobj.weight))
                    self.arguments.add(str(variable))

        _build_regex(domain_rule)
        regex_parts.append('\\|')
        self._trace.append((False, '|'))
        _build_regex(self.is_leaf and self.rule or self.rule.rstrip('/'))
        if not self.is_leaf:
            self._trace.append((False, '/'))

        if self.build_only:
            return
        regex = r'^%s%s$' % (
            u''.join(regex_parts),
            (not self.is_leaf or not self.strict_slashes) and
            '(?<!/)(?P<__suffix__>/?)' or ''
        )
        self._regex = re.compile(regex, re.UNICODE)

這是 werkzurg 框架的 routing 文件中 Rule 類種的一部分的源碼,其中在 def _build_regex(rule): 之前的是一些準備代碼,然後我們接著往下看,for converter, arguments, variable in parse_rule(rule): 這一段代碼,就是 URL 解析,通過調用 parse_rule 函數來實現對我們之前提到的 converter 語法進行解析,緊接著,如果 URL 裡不存在我們 Converter 的語法則 converter 為空,我們執行處理其餘 URL 的邏輯,如果 converter 存在,進行下面的流程,首先,如果我們在 Converter 語法中設定了解析表達式,那麼我們利用 parse_converter_args 函數來處理我們的表達式,方便後續的操作,處理完成後,我們利用 get_converter 方法來初始化我們的 Converter , 代碼如下:


    def get_converter(self, variable_name, converter_name, args, kwargs):
        """Looks up the converter for the given parameter.

        .. versionadded:: 0.9
        """
        if converter_name not in self.map.converters:
            raise LookupError('the converter %r does not exist' % converter_name)
        return self.map.converters[converter_name](self.map, *args, **kwargs)

以我們之前的 demo 為例,

from flask import Flask
from werkzeug.routing import BaseConverter
class RegexConverter(BaseConverter):
    def __init__(self, map, *args):
        self.map = map
        self.regex = args[0]


app = Flask(__name__)
app.url_map.converters['regex'] = RegexConverter

我們已經添加了一個名為 regex 的 Converter 對象,在 get_converter 方法中我們傳入了值為 regexconverter_name 變量,緊接著,我們初始化了一個 RegexConverter 對象的實例,然後返回這個實例

    def compile(self):
        """Compiles the regular expression and stores it."""
        assert self.map is not None, 'rule not bound'

        if self.map.host_matching:
            domain_rule = self.host or ''
        else:
            domain_rule = self.subdomain or ''

        self._trace = []
        self._converters = {}
        self._weights = []
        regex_parts = []

        def _build_regex(rule):
            for converter, arguments, variable in parse_rule(rule):
                if converter is None:
                    regex_parts.append(re.escape(variable))
                    self._trace.append((False, variable))
                    for part in variable.split('/'):
                        if part:
                            self._weights.append((0, -len(part)))
                else:
                    if arguments:
                        c_args, c_kwargs = parse_converter_args(arguments)
                    else:
                        c_args = ()
                        c_kwargs = {}
                    convobj = self.get_converter(
                        variable, converter, c_args, c_kwargs)
############################################################# 無恥分割線
                    regex_parts.append('(?P<%s>%s)' % (variable, convobj.regex))
                    self._converters[variable] = convobj
                    self._trace.append((True, variable))
                    self._weights.append((1, convobj.weight))
                    self.arguments.add(str(variable))

        _build_regex(domain_rule)
        regex_parts.append('\\|')
        self._trace.append((False, '|'))
        _build_regex(self.is_leaf and self.rule or self.rule.rstrip('/'))
        if not self.is_leaf:
            self._trace.append((False, '/'))

        if self.build_only:
            return
        regex = r'^%s%s$' % (
            u''.join(regex_parts),
            (not self.is_leaf or not self.strict_slashes) and
            '(?<!/)(?P<__suffix__>/?)' or ''
        )
        self._regex = re.compile(regex, re.UNICODE)

在分割線後面的代碼中,我們對處理後的 url 進行一些收尾的操作,以我們之前的 demo 為例,我們設定的 /docs/model_utils/<regex(".*"):url> URL 最終轉化成 /docs/model_utils/(?P<url>.*) ,編譯成 re 對象後賦值給 Rule 實例中的 _regex 變量

好了,我們知道處理的部分後,我們大致來看一下怎麼匹配並生成值的吧

    def match(self, path, method=None):
        """Check if the rule matches a given path. Path is a string in the
        form ``"subdomain|/path"`` 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`.

        :internal:
        """
        if not self.build_only:
            m = self._regex.search(path)
            if m is not None:
                groups = m.groupdict()
                # we have a folder like part of the url without a trailing
                # slash and strict slashes enabled. raise an exception that
                # tells the map to redirect to the same url but with a
                # trailing slash
                if self.strict_slashes and not self.is_leaf and \
                        not groups.pop('__suffix__') and \
                        (method is None or self.methods is None or
                         method in self.methods):
                    raise RequestSlash()
                # if we are not in strict slashes mode we have to remove
                # a __suffix__
                elif not self.strict_slashes:
                    del groups['__suffix__']

                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)

                if self.alias and self.map.redirect_defaults:
                    raise RequestAliasRedirect(result)

                return result

這也是 werkzurg 框架的 routing 文件中 Rule 類種的一部分的源碼,在這段代碼中,首先利用 re 對象中的 search 方法,檢測當前傳入的 Path 是否匹配,如果匹配的話,進入後續的處理流程,還記得我們之前最終生成的 /docs/model_utils/(?P<url>.*) 嗎,這裡面利用了正則表達式命名組的語法糖,在這裡,匹配成功後,Python 的 re 庫里給我們提供了一個 groupdict 讓我們取出命名組里所代表的值。然後我們調用 conveter 實例裡面的 to_python 方法來對我們匹配出來的值進行處理(注:這是 Converter 系列對象中的一個可重載方法,我們可以通過重載這個方法,來對我們匹配到的值進行一些邏輯處理,這個我們還是後面再講吧,flag++),然後我們把最終的 result 值返回。

最後的最後,Flask 在獲取 werkzurg 給出的匹配結果後,將匹配的值,放在 request 實例中的 view_args 變量上,最後通過 dispatch_request 對象傳遞給我們的視圖函數,代碼如下

    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`.

        .. versionchanged:: 0.7
           This no longer does the exception handling, this code was
           moved to the new :meth:`full_dispatch_request`.
        """
        req = _request_ctx_stack.top.request
        if req.routing_exception is not None:
            self.raise_routing_exception(req)
        rule = req.url_rule
        # 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
        return self.view_functions[rule.endpoint](**req.view_args)

好了,我們的代碼剖析就到此結束

最後想說幾句#

Flask + Werkzurg 是一套設計實現的非常精妙的組合,不過我們在日常的使用中常常忽略了裡面的美麗的風景,所以這也是我想寫這樣剖析代碼筆記的文章的原因

好了,給老鐵們留幾個思考題,歡迎評論區討論

  • Flask 為什麼不默認支持正則表達式的輸入

  • 诸如 PathConverter 這樣 Werkzurg 內置的 Converter 為什麼在寫表達式的時候可以這樣 /<path:wikipage>/edit 寫,而忽略其中的表達式

  • 前面提到的 parse_converter_args 方法的代碼詳解

好了,就先這樣吧 2333

對了,保佑我文章裡立的 Flag 都能實現(笑)

載入中......
此文章數據所有權由區塊鏈加密技術和智能合約保障僅歸創作者所有。