# 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
    #
    #     http://www.apache.org/licenses/LICENSE-2.0
    #
    # 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
    request:

    .. 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):
                self.connection.write_headers(
                    ResponseStartLine("HTTP/1.1", 200, "OK"),
                    HTTPHeaders({"Content-Length": "2"}),
                    b"OK")
                self.connection.finish()

        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)

                self.finish(resources[path])

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

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

            def find_handler(self, request, **kwargs):
                handler = GetResource if request.method == "GET" else PostResource
                return self.app.get_handler_delegate(request, 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")
            request.finish()

    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("example.com"), 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

    class Router(httputil.HTTPServerConnectionDelegate):
        """Abstract router interface."""

        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)

    class ReversibleRouter(Router):
        """Abstract router interface for routers that can handle named routes
        and support reversing them to original urls.
        """

        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],
        ]
    ]

    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)

        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))

        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(
                        rule.target, request, **target_params
                    )

                    if delegate is not None:
                        return delegate

            return None

        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

    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 rule.name:
                if rule.name in self.named_rules:
                    app_log.warning(
                        "Multiple handlers named %s; replacing previous value", rule.name
                    )
                self.named_rules[rule.name] = 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(rule.target, ReversibleRouter):
                    reversed_url = rule.target.reverse_url(name, *args)
                    if reversed_url is not None:
                        return reversed_url

            return None

    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
            self.target = target
            self.target_kwargs = target_kwargs if target_kwargs else {}
            self.name = 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,
                self.target_kwargs,
                self.name,
            )

    class Matcher(object):
        """Represents a matcher for request features."""

        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()

        def reverse(self, *args: Any) -> Optional[str]:
            """Reconstructs full url from matcher instance and additional arguments."""
            return None

    class AnyMatches(Matcher):
        """Matches any request."""

        def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
            return {}

    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

    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

    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

    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.target
            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,
                self.name,
            )

    @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)