Source code for tornado.routing

# Copyright 2015 The Tornado Authors
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

"""Flexible routing implementation.

Tornado routes HTTP requests to appropriate handlers using `Router`
class implementations. The `tornado.web.Application` class is a
`Router` implementation and may be used directly, or the classes in
this module may be used for additional flexibility. The `RuleRouter`
class can match on more criteria than `.Application`, or the `Router`
interface can be subclassed for maximum customization.

`Router` interface extends `~.httputil.HTTPServerConnectionDelegate`
to provide additional routing capabilities. This also means that any
`Router` implementation can be used directly as a ``request_callback``
for `~.httpserver.HTTPServer` constructor.

`Router` subclass must implement a ``find_handler`` method to provide
a suitable `~.httputil.HTTPMessageDelegate` instance to handle the

.. code-block:: python

    class CustomRouter(Router):
        def find_handler(self, request, **kwargs):
            # some routing logic providing a suitable HTTPMessageDelegate instance
            return MessageDelegate(request.connection)

    class MessageDelegate(HTTPMessageDelegate):
        def __init__(self, connection):
            self.connection = connection

        def finish(self):
                ResponseStartLine("HTTP/1.1", 200, "OK"),
                HTTPHeaders({"Content-Length": "2"}),

    router = CustomRouter()
    server = HTTPServer(router)

The main responsibility of `Router` implementation is to provide a
mapping from a request to `~.httputil.HTTPMessageDelegate` instance
that will handle this request. In the example above we can see that
routing is possible even without instantiating an `~.web.Application`.

For routing to `~.web.RequestHandler` implementations we need an
`~.web.Application` instance. `~.web.Application.get_handler_delegate`
provides a convenient way to create `~.httputil.HTTPMessageDelegate`
for a given request and `~.web.RequestHandler`.

Here is a simple example of how we can we route to
`~.web.RequestHandler` subclasses by HTTP method:

.. code-block:: python

    resources = {}

    class GetResource(RequestHandler):
        def get(self, path):
            if path not in resources:
                raise HTTPError(404)


    class PostResource(RequestHandler):
        def post(self, path):
            resources[path] = self.request.body

    class HTTPMethodRouter(Router):
        def __init__(self, app):
   = app

        def find_handler(self, request, **kwargs):
            handler = GetResource if request.method == "GET" else PostResource
            return, handler, path_args=[request.path])

    router = HTTPMethodRouter(Application())
    server = HTTPServer(router)

`ReversibleRouter` interface adds the ability to distinguish between
the routes and reverse them to the original urls using route's name
and additional arguments. `~.web.Application` is itself an
implementation of `ReversibleRouter` class.

`RuleRouter` and `ReversibleRuleRouter` are implementations of
`Router` and `ReversibleRouter` interfaces and can be used for
creating rule-based routing configurations.

Rules are instances of `Rule` class. They contain a `Matcher`, which
provides the logic for determining whether the rule is a match for a
particular request and a target, which can be one of the following.

1) An instance of `~.httputil.HTTPServerConnectionDelegate`:

.. code-block:: python

    router = RuleRouter([
        Rule(PathMatches("/handler"), ConnectionDelegate()),
        # ... more rules

    class ConnectionDelegate(HTTPServerConnectionDelegate):
        def start_request(self, server_conn, request_conn):
            return MessageDelegate(request_conn)

2) A callable accepting a single argument of `~.httputil.HTTPServerRequest` type:

.. code-block:: python

    router = RuleRouter([
        Rule(PathMatches("/callable"), request_callable)

    def request_callable(request):
        request.write(b"HTTP/1.1 200 OK\\r\\nContent-Length: 2\\r\\n\\r\\nOK")

3) Another `Router` instance:

.. code-block:: python

    router = RuleRouter([
        Rule(PathMatches("/router.*"), CustomRouter())

Of course a nested `RuleRouter` or a `~.web.Application` is allowed:

.. code-block:: python

    router = RuleRouter([
        Rule(HostMatches(""), RuleRouter([
            Rule(PathMatches("/app1/.*"), Application([(r"/app1/handler", Handler)])),

    server = HTTPServer(router)

In the example below `RuleRouter` is used to route between applications:

.. code-block:: python

    app1 = Application([
        (r"/app1/handler", Handler1),
        # other handlers ...

    app2 = Application([
        (r"/app2/handler", Handler2),
        # other handlers ...

    router = RuleRouter([
        Rule(PathMatches("/app1.*"), app1),
        Rule(PathMatches("/app2.*"), app2)

    server = HTTPServer(router)

For more information on application-level routing see docs for `~.web.Application`.

.. versionadded:: 4.5


import re
from functools import partial

from tornado import httputil
from tornado.httpserver import _CallableAdapter
from tornado.escape import url_escape, url_unescape, utf8
from tornado.log import app_log
from tornado.util import basestring_type, import_object, re_unescape, unicode_type

from typing import Any, Union, Optional, Awaitable, List, Dict, Pattern, Tuple, overload

[docs]class Router(httputil.HTTPServerConnectionDelegate): """Abstract router interface."""
[docs] def find_handler( self, request: httputil.HTTPServerRequest, **kwargs: Any ) -> Optional[httputil.HTTPMessageDelegate]: """Must be implemented to return an appropriate instance of `~.httputil.HTTPMessageDelegate` that can serve the request. Routing implementations may pass additional kwargs to extend the routing logic. :arg httputil.HTTPServerRequest request: current HTTP request. :arg kwargs: additional keyword arguments passed by routing implementation. :returns: an instance of `~.httputil.HTTPMessageDelegate` that will be used to process the request. """ raise NotImplementedError()
def start_request( self, server_conn: object, request_conn: httputil.HTTPConnection ) -> httputil.HTTPMessageDelegate: return _RoutingDelegate(self, server_conn, request_conn)
[docs]class ReversibleRouter(Router): """Abstract router interface for routers that can handle named routes and support reversing them to original urls. """
[docs] def reverse_url(self, name: str, *args: Any) -> Optional[str]: """Returns url string for a given route name and arguments or ``None`` if no match is found. :arg str name: route name. :arg args: url parameters. :returns: parametrized url string for a given route name (or ``None``). """ raise NotImplementedError()
class _RoutingDelegate(httputil.HTTPMessageDelegate): def __init__( self, router: Router, server_conn: object, request_conn: httputil.HTTPConnection ) -> None: self.server_conn = server_conn self.request_conn = request_conn self.delegate = None # type: Optional[httputil.HTTPMessageDelegate] self.router = router # type: Router def headers_received( self, start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine], headers: httputil.HTTPHeaders, ) -> Optional[Awaitable[None]]: assert isinstance(start_line, httputil.RequestStartLine) request = httputil.HTTPServerRequest( connection=self.request_conn, server_connection=self.server_conn, start_line=start_line, headers=headers, ) self.delegate = self.router.find_handler(request) if self.delegate is None: app_log.debug( "Delegate for %s %s request not found", start_line.method, start_line.path, ) self.delegate = _DefaultMessageDelegate(self.request_conn) return self.delegate.headers_received(start_line, headers) def data_received(self, chunk: bytes) -> Optional[Awaitable[None]]: assert self.delegate is not None return self.delegate.data_received(chunk) def finish(self) -> None: assert self.delegate is not None self.delegate.finish() def on_connection_close(self) -> None: assert self.delegate is not None self.delegate.on_connection_close() class _DefaultMessageDelegate(httputil.HTTPMessageDelegate): def __init__(self, connection: httputil.HTTPConnection) -> None: self.connection = connection def finish(self) -> None: self.connection.write_headers( httputil.ResponseStartLine("HTTP/1.1", 404, "Not Found"), httputil.HTTPHeaders(), ) self.connection.finish() # _RuleList can either contain pre-constructed Rules or a sequence of # arguments to be passed to the Rule constructor. _RuleList = List[ Union[ "Rule", List[Any], # Can't do detailed typechecking of lists. Tuple[Union[str, "Matcher"], Any], Tuple[Union[str, "Matcher"], Any, Dict[str, Any]], Tuple[Union[str, "Matcher"], Any, Dict[str, Any], str], ] ]
[docs]class RuleRouter(Router): """Rule-based router implementation.""" def __init__(self, rules: Optional[_RuleList] = None) -> None: """Constructs a router from an ordered list of rules:: RuleRouter([ Rule(PathMatches("/handler"), Target), # ... more rules ]) You can also omit explicit `Rule` constructor and use tuples of arguments:: RuleRouter([ (PathMatches("/handler"), Target), ]) `PathMatches` is a default matcher, so the example above can be simplified:: RuleRouter([ ("/handler", Target), ]) In the examples above, ``Target`` can be a nested `Router` instance, an instance of `~.httputil.HTTPServerConnectionDelegate` or an old-style callable, accepting a request argument. :arg rules: a list of `Rule` instances or tuples of `Rule` constructor arguments. """ self.rules = [] # type: List[Rule] if rules: self.add_rules(rules)
[docs] def add_rules(self, rules: _RuleList) -> None: """Appends new rules to the router. :arg rules: a list of Rule instances (or tuples of arguments, which are passed to Rule constructor). """ for rule in rules: if isinstance(rule, (tuple, list)): assert len(rule) in (2, 3, 4) if isinstance(rule[0], basestring_type): rule = Rule(PathMatches(rule[0]), *rule[1:]) else: rule = Rule(*rule) self.rules.append(self.process_rule(rule))
[docs] def process_rule(self, rule: "Rule") -> "Rule": """Override this method for additional preprocessing of each rule. :arg Rule rule: a rule to be processed. :returns: the same or modified Rule instance. """ return rule
def find_handler( self, request: httputil.HTTPServerRequest, **kwargs: Any ) -> Optional[httputil.HTTPMessageDelegate]: for rule in self.rules: target_params = rule.matcher.match(request) if target_params is not None: if rule.target_kwargs: target_params["target_kwargs"] = rule.target_kwargs delegate = self.get_target_delegate(, request, **target_params ) if delegate is not None: return delegate return None
[docs] def get_target_delegate( self, target: Any, request: httputil.HTTPServerRequest, **target_params: Any ) -> Optional[httputil.HTTPMessageDelegate]: """Returns an instance of `~.httputil.HTTPMessageDelegate` for a Rule's target. This method is called by `~.find_handler` and can be extended to provide additional target types. :arg target: a Rule's target. :arg httputil.HTTPServerRequest request: current request. :arg target_params: additional parameters that can be useful for `~.httputil.HTTPMessageDelegate` creation. """ if isinstance(target, Router): return target.find_handler(request, **target_params) elif isinstance(target, httputil.HTTPServerConnectionDelegate): assert request.connection is not None return target.start_request(request.server_connection, request.connection) elif callable(target): assert request.connection is not None return _CallableAdapter( partial(target, **target_params), request.connection ) return None
[docs]class ReversibleRuleRouter(ReversibleRouter, RuleRouter): """A rule-based router that implements ``reverse_url`` method. Each rule added to this router may have a ``name`` attribute that can be used to reconstruct an original uri. The actual reconstruction takes place in a rule's matcher (see `Matcher.reverse`). """ def __init__(self, rules: Optional[_RuleList] = None) -> None: self.named_rules = {} # type: Dict[str, Any] super().__init__(rules) def process_rule(self, rule: "Rule") -> "Rule": rule = super().process_rule(rule) if if in self.named_rules: app_log.warning( "Multiple handlers named %s; replacing previous value", ) self.named_rules[] = rule return rule def reverse_url(self, name: str, *args: Any) -> Optional[str]: if name in self.named_rules: return self.named_rules[name].matcher.reverse(*args) for rule in self.rules: if isinstance(, ReversibleRouter): reversed_url =, *args) if reversed_url is not None: return reversed_url return None
[docs]class Rule(object): """A routing rule.""" def __init__( self, matcher: "Matcher", target: Any, target_kwargs: Optional[Dict[str, Any]] = None, name: Optional[str] = None, ) -> None: """Constructs a Rule instance. :arg Matcher matcher: a `Matcher` instance used for determining whether the rule should be considered a match for a specific request. :arg target: a Rule's target (typically a ``RequestHandler`` or `~.httputil.HTTPServerConnectionDelegate` subclass or even a nested `Router`, depending on routing implementation). :arg dict target_kwargs: a dict of parameters that can be useful at the moment of target instantiation (for example, ``status_code`` for a ``RequestHandler`` subclass). They end up in ``target_params['target_kwargs']`` of `RuleRouter.get_target_delegate` method. :arg str name: the name of the rule that can be used to find it in `ReversibleRouter.reverse_url` implementation. """ if isinstance(target, str): # import the Module and instantiate the class # Must be a fully qualified name (module.ClassName) target = import_object(target) self.matcher = matcher # type: Matcher = target self.target_kwargs = target_kwargs if target_kwargs else {} = name def reverse(self, *args: Any) -> Optional[str]: return self.matcher.reverse(*args) def __repr__(self) -> str: return "%s(%r, %s, kwargs=%r, name=%r)" % ( self.__class__.__name__, self.matcher,, self.target_kwargs,, )
[docs]class Matcher(object): """Represents a matcher for request features."""
[docs] def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]: """Matches current instance against the request. :arg httputil.HTTPServerRequest request: current HTTP request :returns: a dict of parameters to be passed to the target handler (for example, ``handler_kwargs``, ``path_args``, ``path_kwargs`` can be passed for proper `~.web.RequestHandler` instantiation). An empty dict is a valid (and common) return value to indicate a match when the argument-passing features are not used. ``None`` must be returned to indicate that there is no match.""" raise NotImplementedError()
[docs] def reverse(self, *args: Any) -> Optional[str]: """Reconstructs full url from matcher instance and additional arguments.""" return None
[docs]class AnyMatches(Matcher): """Matches any request.""" def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]: return {}
[docs]class HostMatches(Matcher): """Matches requests from hosts specified by ``host_pattern`` regex.""" def __init__(self, host_pattern: Union[str, Pattern]) -> None: if isinstance(host_pattern, basestring_type): if not host_pattern.endswith("$"): host_pattern += "$" self.host_pattern = re.compile(host_pattern) else: self.host_pattern = host_pattern def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]: if self.host_pattern.match(request.host_name): return {} return None
[docs]class DefaultHostMatches(Matcher): """Matches requests from host that is equal to application's default_host. Always returns no match if ``X-Real-Ip`` header is present. """ def __init__(self, application: Any, host_pattern: Pattern) -> None: self.application = application self.host_pattern = host_pattern def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]: # Look for default host if not behind load balancer (for debugging) if "X-Real-Ip" not in request.headers: if self.host_pattern.match(self.application.default_host): return {} return None
[docs]class PathMatches(Matcher): """Matches requests with paths specified by ``path_pattern`` regex.""" def __init__(self, path_pattern: Union[str, Pattern]) -> None: if isinstance(path_pattern, basestring_type): if not path_pattern.endswith("$"): path_pattern += "$" self.regex = re.compile(path_pattern) else: self.regex = path_pattern assert len(self.regex.groupindex) in (0, self.regex.groups), ( "groups in url regexes must either be all named or all " "positional: %r" % self.regex.pattern ) self._path, self._group_count = self._find_groups() def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]: match = self.regex.match(request.path) if match is None: return None if not self.regex.groups: return {} path_args = [] # type: List[bytes] path_kwargs = {} # type: Dict[str, bytes] # Pass matched groups to the handler. Since # match.groups() includes both named and # unnamed groups, we want to use either groups # or groupdict but not both. if self.regex.groupindex: path_kwargs = dict( (str(k), _unquote_or_none(v)) for (k, v) in match.groupdict().items() ) else: path_args = [_unquote_or_none(s) for s in match.groups()] return dict(path_args=path_args, path_kwargs=path_kwargs) def reverse(self, *args: Any) -> Optional[str]: if self._path is None: raise ValueError("Cannot reverse url regex " + self.regex.pattern) assert len(args) == self._group_count, ( "required number of arguments " "not found" ) if not len(args): return self._path converted_args = [] for a in args: if not isinstance(a, (unicode_type, bytes)): a = str(a) converted_args.append(url_escape(utf8(a), plus=False)) return self._path % tuple(converted_args) def _find_groups(self) -> Tuple[Optional[str], Optional[int]]: """Returns a tuple (reverse string, group count) for a url. For example: Given the url pattern /([0-9]{4})/([a-z-]+)/, this method would return ('/%s/%s/', 2). """ pattern = self.regex.pattern if pattern.startswith("^"): pattern = pattern[1:] if pattern.endswith("$"): pattern = pattern[:-1] if self.regex.groups != pattern.count("("): # The pattern is too complicated for our simplistic matching, # so we can't support reversing it. return None, None pieces = [] for fragment in pattern.split("("): if ")" in fragment: paren_loc = fragment.index(")") if paren_loc >= 0: try: unescaped_fragment = re_unescape(fragment[paren_loc + 1 :]) except ValueError: # If we can't unescape part of it, we can't # reverse this url. return (None, None) pieces.append("%s" + unescaped_fragment) else: try: unescaped_fragment = re_unescape(fragment) except ValueError: # If we can't unescape part of it, we can't # reverse this url. return (None, None) pieces.append(unescaped_fragment) return "".join(pieces), self.regex.groups
[docs]class URLSpec(Rule): """Specifies mappings between URLs and handlers. .. versionchanged: 4.5 `URLSpec` is now a subclass of a `Rule` with `PathMatches` matcher and is preserved for backwards compatibility. """ def __init__( self, pattern: Union[str, Pattern], handler: Any, kwargs: Optional[Dict[str, Any]] = None, name: Optional[str] = None, ) -> None: """Parameters: * ``pattern``: Regular expression to be matched. Any capturing groups in the regex will be passed in to the handler's get/post/etc methods as arguments (by keyword if named, by position if unnamed. Named and unnamed capturing groups may not be mixed in the same rule). * ``handler``: `~.web.RequestHandler` subclass to be invoked. * ``kwargs`` (optional): A dictionary of additional arguments to be passed to the handler's constructor. * ``name`` (optional): A name for this handler. Used by `~.web.Application.reverse_url`. """ matcher = PathMatches(pattern) super().__init__(matcher, handler, kwargs, name) self.regex = matcher.regex self.handler_class = self.kwargs = kwargs def __repr__(self) -> str: return "%s(%r, %s, kwargs=%r, name=%r)" % ( self.__class__.__name__, self.regex.pattern, self.handler_class, self.kwargs,, )
@overload def _unquote_or_none(s: str) -> bytes: pass @overload # noqa: F811 def _unquote_or_none(s: None) -> None: pass def _unquote_or_none(s: Optional[str]) -> Optional[bytes]: # noqa: F811 """None-safe wrapper around url_unescape to handle unmatched optional groups correctly. Note that args are passed as bytes so the handler can decide what encoding to use. """ if s is None: return s return url_unescape(s, encoding=None, plus=False)