Flask源码剖析(3): 路由

1
2
3
@app.route('/')
def hello_world():
    return 'Hello World!'

1
2
3
4
def hello():
    return "hello, world!"

app.add_url_rule('/', 'hello', hello)

1
2
3
4
url_map = Map([
    Rule('/', endpoint='hello')
])

1
2
3
4
5
6
7
8
9
def route(self, rule, **options):
    """Like :meth:`Flask.route` but for a blueprint.  The endpoint for the
    :func:`url_for` function is prefixed with the name of the blueprint.
    """
    def decorator(f):
        endpoint = options.pop("endpoint", f.__name__)
        self.add_url_rule(rule, endpoint, f, **options)
        return f
    return decorator

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@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 methods is None:
        methods = getattr(view_func, 'methods', None) or ('GET',)
    methods = set(methods)

    required_methods = set(getattr(view_func, 'required_methods', ()))
    
    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

    methods |= required_methods

    options['defaults'] = options.get('defaults') or None

    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

1
2
3
4
5
6
7
8
9
    def dispatch_request(self):
        req = _request_ctx_stack.top.request
        if req.routing_exception is not None:
            self.raise_routing_exception(req)
        rule = req.url_rule
        if getattr(rule, 'provide_automatic_options', False) \
           and req.method == 'OPTIONS':
            return self.make_default_options_response()
        return self.view_functions[rule.endpoint](**req.view_args)

1
2
3
4
5
6
7
8
9
10
11
    class Flask(_PackageBoundObject):
        def create_url_adapter(self, request):
            if request is not None:
                return self.url_map.bind_to_environ(request.environ,
                    server_name=self.config['SERVER_NAME'])
            if self.config['SERVER_NAME'] is not None:
                return self.url_map.bind(
                    self.config['SERVER_NAME'],
                    script_name=self.config['APPLICATION_ROOT'] or '/',
                    url_scheme=self.config['PREFERRED_URL_SCHEME'])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
    class RequestContext(object):
        def __init__(self, app, environ, request=None):
            self.app = app
            if request is None:
                request = app.request_class(environ)
            self.request = request
            self.url_adapter = app.create_url_adapter(self.request)
    
            self.match_request()
            
        def match_request(self):
            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
                
        def push(self):
            top = _request_ctx_stack.top
            if top is not None and top.preserved:
                top.pop(top._preserved_exc)
    
            app_ctx = _app_ctx_stack.top
            if app_ctx is None or app_ctx.app != self.app:
                app_ctx = self.app.app_context()
                app_ctx.push()
                self._implicit_app_ctx_stack.append(app_ctx)
            else:
                self._implicit_app_ctx_stack.append(None)
    
            if hasattr(sys, 'exc_clear'):
                sys.exc_clear()
    
            _request_ctx_stack.push(self)
    
            self.session = self.app.open_session(self.request)
            if self.session is None:
                self.session = self.app.make_null_session()
                
        def __enter__(self):
            self.push()
            return self
    
        def __exit__(self, exc_type, exc_value, tb):
            self.auto_pop(exc_value)
    
            if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
                reraise(exc_type, exc_value, tb)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
    def match(self, path):
        if not self.build_only:
            m = self._regex.search(path)
            if m is not None:
                groups = m.groupdict()
                if self.strict_slashes and not self.is_leaf and \
                   not groups.pop('__suffix__'):
                    raise RequestSlash()
                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