Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/routing.py: 28%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2015 The Tornado Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may
4# not use this file except in compliance with the License. You may obtain
5# a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations
13# under the License.
15"""Flexible routing implementation.
17Tornado routes HTTP requests to appropriate handlers using `Router`
18class implementations. The `tornado.web.Application` class is a
19`Router` implementation and may be used directly, or the classes in
20this module may be used for additional flexibility. The `RuleRouter`
21class can match on more criteria than `.Application`, or the `Router`
22interface can be subclassed for maximum customization.
24`Router` interface extends `~.httputil.HTTPServerConnectionDelegate`
25to provide additional routing capabilities. This also means that any
26`Router` implementation can be used directly as a ``request_callback``
27for `~.httpserver.HTTPServer` constructor.
29`Router` subclass must implement a ``find_handler`` method to provide
30a suitable `~.httputil.HTTPMessageDelegate` instance to handle the
31request:
33.. code-block:: python
35 class CustomRouter(Router):
36 def find_handler(self, request, **kwargs):
37 # some routing logic providing a suitable HTTPMessageDelegate instance
38 return MessageDelegate(request.connection)
40 class MessageDelegate(HTTPMessageDelegate):
41 def __init__(self, connection):
42 self.connection = connection
44 def finish(self):
45 self.connection.write_headers(
46 ResponseStartLine("HTTP/1.1", 200, "OK"),
47 HTTPHeaders({"Content-Length": "2"}),
48 b"OK")
49 self.connection.finish()
51 router = CustomRouter()
52 server = HTTPServer(router)
54The main responsibility of `Router` implementation is to provide a
55mapping from a request to `~.httputil.HTTPMessageDelegate` instance
56that will handle this request. In the example above we can see that
57routing is possible even without instantiating an `~.web.Application`.
59For routing to `~.web.RequestHandler` implementations we need an
60`~.web.Application` instance. `~.web.Application.get_handler_delegate`
61provides a convenient way to create `~.httputil.HTTPMessageDelegate`
62for a given request and `~.web.RequestHandler`.
64Here is a simple example of how we can we route to
65`~.web.RequestHandler` subclasses by HTTP method:
67.. code-block:: python
69 resources = {}
71 class GetResource(RequestHandler):
72 def get(self, path):
73 if path not in resources:
74 raise HTTPError(404)
76 self.finish(resources[path])
78 class PostResource(RequestHandler):
79 def post(self, path):
80 resources[path] = self.request.body
82 class HTTPMethodRouter(Router):
83 def __init__(self, app):
84 self.app = app
86 def find_handler(self, request, **kwargs):
87 handler = GetResource if request.method == "GET" else PostResource
88 return self.app.get_handler_delegate(request, handler, path_args=[request.path])
90 router = HTTPMethodRouter(Application())
91 server = HTTPServer(router)
93`ReversibleRouter` interface adds the ability to distinguish between
94the routes and reverse them to the original urls using route's name
95and additional arguments. `~.web.Application` is itself an
96implementation of `ReversibleRouter` class.
98`RuleRouter` and `ReversibleRuleRouter` are implementations of
99`Router` and `ReversibleRouter` interfaces and can be used for
100creating rule-based routing configurations.
102Rules are instances of `Rule` class. They contain a `Matcher`, which
103provides the logic for determining whether the rule is a match for a
104particular request and a target, which can be one of the following.
1061) An instance of `~.httputil.HTTPServerConnectionDelegate`:
108.. code-block:: python
110 router = RuleRouter([
111 Rule(PathMatches("/handler"), ConnectionDelegate()),
112 # ... more rules
113 ])
115 class ConnectionDelegate(HTTPServerConnectionDelegate):
116 def start_request(self, server_conn, request_conn):
117 return MessageDelegate(request_conn)
1192) A callable accepting a single argument of `~.httputil.HTTPServerRequest` type:
121.. code-block:: python
123 router = RuleRouter([
124 Rule(PathMatches("/callable"), request_callable)
125 ])
127 def request_callable(request):
128 request.write(b"HTTP/1.1 200 OK\\r\\nContent-Length: 2\\r\\n\\r\\nOK")
129 request.finish()
1313) Another `Router` instance:
133.. code-block:: python
135 router = RuleRouter([
136 Rule(PathMatches("/router.*"), CustomRouter())
137 ])
139Of course a nested `RuleRouter` or a `~.web.Application` is allowed:
141.. code-block:: python
143 router = RuleRouter([
144 Rule(HostMatches("example.com"), RuleRouter([
145 Rule(PathMatches("/app1/.*"), Application([(r"/app1/handler", Handler)])),
146 ]))
147 ])
149 server = HTTPServer(router)
151In the example below `RuleRouter` is used to route between applications:
153.. code-block:: python
155 app1 = Application([
156 (r"/app1/handler", Handler1),
157 # other handlers ...
158 ])
160 app2 = Application([
161 (r"/app2/handler", Handler2),
162 # other handlers ...
163 ])
165 router = RuleRouter([
166 Rule(PathMatches("/app1.*"), app1),
167 Rule(PathMatches("/app2.*"), app2)
168 ])
170 server = HTTPServer(router)
172For more information on application-level routing see docs for `~.web.Application`.
174.. versionadded:: 4.5
176"""
178import re
179from functools import partial
181from tornado import httputil
182from tornado.httpserver import _CallableAdapter
183from tornado.escape import url_escape, url_unescape, utf8
184from tornado.log import app_log
185from tornado.util import basestring_type, import_object, re_unescape, unicode_type
187from typing import (
188 Any,
189 Union,
190 Optional,
191 Awaitable,
192 List,
193 Dict,
194 Pattern,
195 Tuple,
196 overload,
197 Sequence,
198)
201class Router(httputil.HTTPServerConnectionDelegate):
202 """Abstract router interface."""
204 def find_handler(
205 self, request: httputil.HTTPServerRequest, **kwargs: Any
206 ) -> Optional[httputil.HTTPMessageDelegate]:
207 """Must be implemented to return an appropriate instance of `~.httputil.HTTPMessageDelegate`
208 that can serve the request.
209 Routing implementations may pass additional kwargs to extend the routing logic.
211 :arg httputil.HTTPServerRequest request: current HTTP request.
212 :arg kwargs: additional keyword arguments passed by routing implementation.
213 :returns: an instance of `~.httputil.HTTPMessageDelegate` that will be used to
214 process the request.
215 """
216 raise NotImplementedError()
218 def start_request(
219 self, server_conn: object, request_conn: httputil.HTTPConnection
220 ) -> httputil.HTTPMessageDelegate:
221 return _RoutingDelegate(self, server_conn, request_conn)
224class ReversibleRouter(Router):
225 """Abstract router interface for routers that can handle named routes
226 and support reversing them to original urls.
227 """
229 def reverse_url(self, name: str, *args: Any) -> Optional[str]:
230 """Returns url string for a given route name and arguments
231 or ``None`` if no match is found.
233 :arg str name: route name.
234 :arg args: url parameters.
235 :returns: parametrized url string for a given route name (or ``None``).
236 """
237 raise NotImplementedError()
240class _RoutingDelegate(httputil.HTTPMessageDelegate):
241 def __init__(
242 self, router: Router, server_conn: object, request_conn: httputil.HTTPConnection
243 ) -> None:
244 self.server_conn = server_conn
245 self.request_conn = request_conn
246 self.delegate = None # type: Optional[httputil.HTTPMessageDelegate]
247 self.router = router # type: Router
249 def headers_received(
250 self,
251 start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine],
252 headers: httputil.HTTPHeaders,
253 ) -> Optional[Awaitable[None]]:
254 assert isinstance(start_line, httputil.RequestStartLine)
255 request = httputil.HTTPServerRequest(
256 connection=self.request_conn,
257 server_connection=self.server_conn,
258 start_line=start_line,
259 headers=headers,
260 )
262 self.delegate = self.router.find_handler(request)
263 if self.delegate is None:
264 app_log.debug(
265 "Delegate for %s %s request not found",
266 start_line.method,
267 start_line.path,
268 )
269 self.delegate = _DefaultMessageDelegate(self.request_conn)
271 return self.delegate.headers_received(start_line, headers)
273 def data_received(self, chunk: bytes) -> Optional[Awaitable[None]]:
274 assert self.delegate is not None
275 return self.delegate.data_received(chunk)
277 def finish(self) -> None:
278 assert self.delegate is not None
279 self.delegate.finish()
281 def on_connection_close(self) -> None:
282 assert self.delegate is not None
283 self.delegate.on_connection_close()
286class _DefaultMessageDelegate(httputil.HTTPMessageDelegate):
287 def __init__(self, connection: httputil.HTTPConnection) -> None:
288 self.connection = connection
290 def finish(self) -> None:
291 self.connection.write_headers(
292 httputil.ResponseStartLine("HTTP/1.1", 404, "Not Found"),
293 httputil.HTTPHeaders(),
294 )
295 self.connection.finish()
298# _RuleList can either contain pre-constructed Rules or a sequence of
299# arguments to be passed to the Rule constructor.
300_RuleList = Sequence[
301 Union[
302 "Rule",
303 List[Any], # Can't do detailed typechecking of lists.
304 Tuple[Union[str, "Matcher"], Any],
305 Tuple[Union[str, "Matcher"], Any, Dict[str, Any]],
306 Tuple[Union[str, "Matcher"], Any, Dict[str, Any], str],
307 ]
308]
311class RuleRouter(Router):
312 """Rule-based router implementation."""
314 def __init__(self, rules: Optional[_RuleList] = None) -> None:
315 """Constructs a router from an ordered list of rules::
317 RuleRouter([
318 Rule(PathMatches("/handler"), Target),
319 # ... more rules
320 ])
322 You can also omit explicit `Rule` constructor and use tuples of arguments::
324 RuleRouter([
325 (PathMatches("/handler"), Target),
326 ])
328 `PathMatches` is a default matcher, so the example above can be simplified::
330 RuleRouter([
331 ("/handler", Target),
332 ])
334 In the examples above, ``Target`` can be a nested `Router` instance, an instance of
335 `~.httputil.HTTPServerConnectionDelegate` or an old-style callable,
336 accepting a request argument.
338 :arg rules: a list of `Rule` instances or tuples of `Rule`
339 constructor arguments.
340 """
341 self.rules = [] # type: List[Rule]
342 if rules:
343 self.add_rules(rules)
345 def add_rules(self, rules: _RuleList) -> None:
346 """Appends new rules to the router.
348 :arg rules: a list of Rule instances (or tuples of arguments, which are
349 passed to Rule constructor).
350 """
351 for rule in rules:
352 if isinstance(rule, (tuple, list)):
353 assert len(rule) in (2, 3, 4)
354 if isinstance(rule[0], basestring_type):
355 rule = Rule(PathMatches(rule[0]), *rule[1:])
356 else:
357 rule = Rule(*rule)
359 self.rules.append(self.process_rule(rule))
361 def process_rule(self, rule: "Rule") -> "Rule":
362 """Override this method for additional preprocessing of each rule.
364 :arg Rule rule: a rule to be processed.
365 :returns: the same or modified Rule instance.
366 """
367 return rule
369 def find_handler(
370 self, request: httputil.HTTPServerRequest, **kwargs: Any
371 ) -> Optional[httputil.HTTPMessageDelegate]:
372 for rule in self.rules:
373 target_params = rule.matcher.match(request)
374 if target_params is not None:
375 if rule.target_kwargs:
376 target_params["target_kwargs"] = rule.target_kwargs
378 delegate = self.get_target_delegate(
379 rule.target, request, **target_params
380 )
382 if delegate is not None:
383 return delegate
385 return None
387 def get_target_delegate(
388 self, target: Any, request: httputil.HTTPServerRequest, **target_params: Any
389 ) -> Optional[httputil.HTTPMessageDelegate]:
390 """Returns an instance of `~.httputil.HTTPMessageDelegate` for a
391 Rule's target. This method is called by `~.find_handler` and can be
392 extended to provide additional target types.
394 :arg target: a Rule's target.
395 :arg httputil.HTTPServerRequest request: current request.
396 :arg target_params: additional parameters that can be useful
397 for `~.httputil.HTTPMessageDelegate` creation.
398 """
399 if isinstance(target, Router):
400 return target.find_handler(request, **target_params)
402 elif isinstance(target, httputil.HTTPServerConnectionDelegate):
403 assert request.connection is not None
404 return target.start_request(request.server_connection, request.connection)
406 elif callable(target):
407 assert request.connection is not None
408 return _CallableAdapter(
409 partial(target, **target_params), request.connection
410 )
412 return None
415class ReversibleRuleRouter(ReversibleRouter, RuleRouter):
416 """A rule-based router that implements ``reverse_url`` method.
418 Each rule added to this router may have a ``name`` attribute that can be
419 used to reconstruct an original uri. The actual reconstruction takes place
420 in a rule's matcher (see `Matcher.reverse`).
421 """
423 def __init__(self, rules: Optional[_RuleList] = None) -> None:
424 self.named_rules = {} # type: Dict[str, Any]
425 super().__init__(rules)
427 def process_rule(self, rule: "Rule") -> "Rule":
428 rule = super().process_rule(rule)
430 if rule.name:
431 if rule.name in self.named_rules:
432 app_log.warning(
433 "Multiple handlers named %s; replacing previous value", rule.name
434 )
435 self.named_rules[rule.name] = rule
437 return rule
439 def reverse_url(self, name: str, *args: Any) -> Optional[str]:
440 if name in self.named_rules:
441 return self.named_rules[name].matcher.reverse(*args)
443 for rule in self.rules:
444 if isinstance(rule.target, ReversibleRouter):
445 reversed_url = rule.target.reverse_url(name, *args)
446 if reversed_url is not None:
447 return reversed_url
449 return None
452class Rule:
453 """A routing rule."""
455 def __init__(
456 self,
457 matcher: "Matcher",
458 target: Any,
459 target_kwargs: Optional[Dict[str, Any]] = None,
460 name: Optional[str] = None,
461 ) -> None:
462 """Constructs a Rule instance.
464 :arg Matcher matcher: a `Matcher` instance used for determining
465 whether the rule should be considered a match for a specific
466 request.
467 :arg target: a Rule's target (typically a ``RequestHandler`` or
468 `~.httputil.HTTPServerConnectionDelegate` subclass or even a nested `Router`,
469 depending on routing implementation).
470 :arg dict target_kwargs: a dict of parameters that can be useful
471 at the moment of target instantiation (for example, ``status_code``
472 for a ``RequestHandler`` subclass). They end up in
473 ``target_params['target_kwargs']`` of `RuleRouter.get_target_delegate`
474 method.
475 :arg str name: the name of the rule that can be used to find it
476 in `ReversibleRouter.reverse_url` implementation.
477 """
478 if isinstance(target, str):
479 # import the Module and instantiate the class
480 # Must be a fully qualified name (module.ClassName)
481 target = import_object(target)
483 self.matcher = matcher # type: Matcher
484 self.target = target
485 self.target_kwargs = target_kwargs if target_kwargs else {}
486 self.name = name
488 def reverse(self, *args: Any) -> Optional[str]:
489 return self.matcher.reverse(*args)
491 def __repr__(self) -> str:
492 return "{}({!r}, {}, kwargs={!r}, name={!r})".format(
493 self.__class__.__name__,
494 self.matcher,
495 self.target,
496 self.target_kwargs,
497 self.name,
498 )
501class Matcher:
502 """Represents a matcher for request features."""
504 def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
505 """Matches current instance against the request.
507 :arg httputil.HTTPServerRequest request: current HTTP request
508 :returns: a dict of parameters to be passed to the target handler
509 (for example, ``handler_kwargs``, ``path_args``, ``path_kwargs``
510 can be passed for proper `~.web.RequestHandler` instantiation).
511 An empty dict is a valid (and common) return value to indicate a match
512 when the argument-passing features are not used.
513 ``None`` must be returned to indicate that there is no match."""
514 raise NotImplementedError()
516 def reverse(self, *args: Any) -> Optional[str]:
517 """Reconstructs full url from matcher instance and additional arguments."""
518 return None
521class AnyMatches(Matcher):
522 """Matches any request."""
524 def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
525 return {}
528class HostMatches(Matcher):
529 """Matches requests from hosts specified by ``host_pattern`` regex."""
531 def __init__(self, host_pattern: Union[str, Pattern]) -> None:
532 if isinstance(host_pattern, basestring_type):
533 if not host_pattern.endswith("$"):
534 host_pattern += "$"
535 self.host_pattern = re.compile(host_pattern)
536 else:
537 self.host_pattern = host_pattern
539 def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
540 if self.host_pattern.match(request.host_name):
541 return {}
543 return None
546class DefaultHostMatches(Matcher):
547 """Matches requests from host that is equal to application's default_host.
548 Always returns no match if ``X-Real-Ip`` header is present.
549 """
551 def __init__(self, application: Any, host_pattern: Pattern) -> None:
552 self.application = application
553 self.host_pattern = host_pattern
555 def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
556 # Look for default host if not behind load balancer (for debugging)
557 if "X-Real-Ip" not in request.headers:
558 if self.host_pattern.match(self.application.default_host):
559 return {}
560 return None
563class PathMatches(Matcher):
564 """Matches requests with paths specified by ``path_pattern`` regex."""
566 def __init__(self, path_pattern: Union[str, Pattern]) -> None:
567 if isinstance(path_pattern, basestring_type):
568 if not path_pattern.endswith("$"):
569 path_pattern += "$"
570 self.regex = re.compile(path_pattern)
571 else:
572 self.regex = path_pattern
574 assert len(self.regex.groupindex) in (0, self.regex.groups), (
575 "groups in url regexes must either be all named or all "
576 "positional: %r" % self.regex.pattern
577 )
579 self._path, self._group_count = self._find_groups()
581 def match(self, request: httputil.HTTPServerRequest) -> Optional[Dict[str, Any]]:
582 match = self.regex.match(request.path)
583 if match is None:
584 return None
585 if not self.regex.groups:
586 return {}
588 path_args = [] # type: List[bytes]
589 path_kwargs = {} # type: Dict[str, bytes]
591 # Pass matched groups to the handler. Since
592 # match.groups() includes both named and
593 # unnamed groups, we want to use either groups
594 # or groupdict but not both.
595 if self.regex.groupindex:
596 path_kwargs = {
597 str(k): _unquote_or_none(v) for (k, v) in match.groupdict().items()
598 }
599 else:
600 path_args = [_unquote_or_none(s) for s in match.groups()]
602 return dict(path_args=path_args, path_kwargs=path_kwargs)
604 def reverse(self, *args: Any) -> Optional[str]:
605 if self._path is None:
606 raise ValueError("Cannot reverse url regex " + self.regex.pattern)
607 assert len(args) == self._group_count, (
608 "required number of arguments " "not found"
609 )
610 if not len(args):
611 return self._path
612 converted_args = []
613 for a in args:
614 if not isinstance(a, (unicode_type, bytes)):
615 a = str(a)
616 converted_args.append(url_escape(utf8(a), plus=False))
617 return self._path % tuple(converted_args)
619 def _find_groups(self) -> Tuple[Optional[str], Optional[int]]:
620 """Returns a tuple (reverse string, group count) for a url.
622 For example: Given the url pattern /([0-9]{4})/([a-z-]+)/, this method
623 would return ('/%s/%s/', 2).
624 """
625 pattern = self.regex.pattern
626 if pattern.startswith("^"):
627 pattern = pattern[1:]
628 if pattern.endswith("$"):
629 pattern = pattern[:-1]
631 if self.regex.groups != pattern.count("("):
632 # The pattern is too complicated for our simplistic matching,
633 # so we can't support reversing it.
634 return None, None
636 pieces = []
637 for fragment in pattern.split("("):
638 if ")" in fragment:
639 paren_loc = fragment.index(")")
640 if paren_loc >= 0:
641 try:
642 unescaped_fragment = re_unescape(fragment[paren_loc + 1 :])
643 except ValueError:
644 # If we can't unescape part of it, we can't
645 # reverse this url.
646 return (None, None)
647 pieces.append("%s" + unescaped_fragment)
648 else:
649 try:
650 unescaped_fragment = re_unescape(fragment)
651 except ValueError:
652 # If we can't unescape part of it, we can't
653 # reverse this url.
654 return (None, None)
655 pieces.append(unescaped_fragment)
657 return "".join(pieces), self.regex.groups
660class URLSpec(Rule):
661 """Specifies mappings between URLs and handlers.
663 .. versionchanged: 4.5
664 `URLSpec` is now a subclass of a `Rule` with `PathMatches` matcher and is preserved for
665 backwards compatibility.
666 """
668 def __init__(
669 self,
670 pattern: Union[str, Pattern],
671 handler: Any,
672 kwargs: Optional[Dict[str, Any]] = None,
673 name: Optional[str] = None,
674 ) -> None:
675 """Parameters:
677 * ``pattern``: Regular expression to be matched. Any capturing
678 groups in the regex will be passed in to the handler's
679 get/post/etc methods as arguments (by keyword if named, by
680 position if unnamed. Named and unnamed capturing groups
681 may not be mixed in the same rule).
683 * ``handler``: `~.web.RequestHandler` subclass to be invoked.
685 * ``kwargs`` (optional): A dictionary of additional arguments
686 to be passed to the handler's constructor.
688 * ``name`` (optional): A name for this handler. Used by
689 `~.web.Application.reverse_url`.
691 """
692 matcher = PathMatches(pattern)
693 super().__init__(matcher, handler, kwargs, name)
695 self.regex = matcher.regex
696 self.handler_class = self.target
697 self.kwargs = kwargs
699 def __repr__(self) -> str:
700 return "{}({!r}, {}, kwargs={!r}, name={!r})".format(
701 self.__class__.__name__,
702 self.regex.pattern,
703 self.handler_class,
704 self.kwargs,
705 self.name,
706 )
709@overload
710def _unquote_or_none(s: str) -> bytes:
711 pass
714@overload # noqa: F811
715def _unquote_or_none(s: None) -> None:
716 pass
719def _unquote_or_none(s: Optional[str]) -> Optional[bytes]: # noqa: F811
720 """None-safe wrapper around url_unescape to handle unmatched optional
721 groups correctly.
723 Note that args are passed as bytes so the handler can decide what
724 encoding to use.
725 """
726 if s is None:
727 return s
728 return url_unescape(s, encoding=None, plus=False)