Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| import typing as t | |
| import warnings | |
| from pprint import pformat | |
| from threading import Lock | |
| from urllib.parse import quote | |
| from urllib.parse import urljoin | |
| from urllib.parse import urlunsplit | |
| from .._internal import _get_environ | |
| from .._internal import _wsgi_decoding_dance | |
| from ..datastructures import ImmutableDict | |
| from ..datastructures import MultiDict | |
| from ..exceptions import BadHost | |
| from ..exceptions import HTTPException | |
| from ..exceptions import MethodNotAllowed | |
| from ..exceptions import NotFound | |
| from ..urls import _urlencode | |
| from ..wsgi import get_host | |
| from .converters import DEFAULT_CONVERTERS | |
| from .exceptions import BuildError | |
| from .exceptions import NoMatch | |
| from .exceptions import RequestAliasRedirect | |
| from .exceptions import RequestPath | |
| from .exceptions import RequestRedirect | |
| from .exceptions import WebsocketMismatch | |
| from .matcher import StateMachineMatcher | |
| from .rules import _simple_rule_re | |
| from .rules import Rule | |
| if t.TYPE_CHECKING: | |
| from _typeshed.wsgi import WSGIApplication | |
| from _typeshed.wsgi import WSGIEnvironment | |
| from ..wrappers.request import Request | |
| from .converters import BaseConverter | |
| from .rules import RuleFactory | |
| class Map: | |
| """The map class stores all the URL rules and some configuration | |
| parameters. Some of the configuration values are only stored on the | |
| `Map` instance since those affect all rules, others are just defaults | |
| and can be overridden for each rule. Note that you have to specify all | |
| arguments besides the `rules` as keyword arguments! | |
| :param rules: sequence of url rules for this map. | |
| :param default_subdomain: The default subdomain for rules without a | |
| subdomain defined. | |
| :param strict_slashes: If a rule ends with a slash but the matched | |
| URL does not, redirect to the URL with a trailing slash. | |
| :param merge_slashes: Merge consecutive slashes when matching or | |
| building URLs. Matches will redirect to the normalized URL. | |
| Slashes in variable parts are not merged. | |
| :param redirect_defaults: This will redirect to the default rule if it | |
| wasn't visited that way. This helps creating | |
| unique URLs. | |
| :param converters: A dict of converters that adds additional converters | |
| to the list of converters. If you redefine one | |
| converter this will override the original one. | |
| :param sort_parameters: If set to `True` the url parameters are sorted. | |
| See `url_encode` for more details. | |
| :param sort_key: The sort key function for `url_encode`. | |
| :param host_matching: if set to `True` it enables the host matching | |
| feature and disables the subdomain one. If | |
| enabled the `host` parameter to rules is used | |
| instead of the `subdomain` one. | |
| .. versionchanged:: 3.0 | |
| The ``charset`` and ``encoding_errors`` parameters were removed. | |
| .. versionchanged:: 1.0 | |
| If ``url_scheme`` is ``ws`` or ``wss``, only WebSocket rules will match. | |
| .. versionchanged:: 1.0 | |
| The ``merge_slashes`` parameter was added. | |
| .. versionchanged:: 0.7 | |
| The ``encoding_errors`` and ``host_matching`` parameters were added. | |
| .. versionchanged:: 0.5 | |
| The ``sort_parameters`` and ``sort_key`` paramters were added. | |
| """ | |
| #: A dict of default converters to be used. | |
| default_converters = ImmutableDict(DEFAULT_CONVERTERS) | |
| #: The type of lock to use when updating. | |
| #: | |
| #: .. versionadded:: 1.0 | |
| lock_class = Lock | |
| def __init__( | |
| self, | |
| rules: t.Iterable[RuleFactory] | None = None, | |
| default_subdomain: str = "", | |
| strict_slashes: bool = True, | |
| merge_slashes: bool = True, | |
| redirect_defaults: bool = True, | |
| converters: t.Mapping[str, type[BaseConverter]] | None = None, | |
| sort_parameters: bool = False, | |
| sort_key: t.Callable[[t.Any], t.Any] | None = None, | |
| host_matching: bool = False, | |
| ) -> None: | |
| self._matcher = StateMachineMatcher(merge_slashes) | |
| self._rules_by_endpoint: dict[t.Any, list[Rule]] = {} | |
| self._remap = True | |
| self._remap_lock = self.lock_class() | |
| self.default_subdomain = default_subdomain | |
| self.strict_slashes = strict_slashes | |
| self.redirect_defaults = redirect_defaults | |
| self.host_matching = host_matching | |
| self.converters = self.default_converters.copy() | |
| if converters: | |
| self.converters.update(converters) | |
| self.sort_parameters = sort_parameters | |
| self.sort_key = sort_key | |
| for rulefactory in rules or (): | |
| self.add(rulefactory) | |
| def merge_slashes(self) -> bool: | |
| return self._matcher.merge_slashes | |
| def merge_slashes(self, value: bool) -> None: | |
| self._matcher.merge_slashes = value | |
| def is_endpoint_expecting(self, endpoint: t.Any, *arguments: str) -> bool: | |
| """Iterate over all rules and check if the endpoint expects | |
| the arguments provided. This is for example useful if you have | |
| some URLs that expect a language code and others that do not and | |
| you want to wrap the builder a bit so that the current language | |
| code is automatically added if not provided but endpoints expect | |
| it. | |
| :param endpoint: the endpoint to check. | |
| :param arguments: this function accepts one or more arguments | |
| as positional arguments. Each one of them is | |
| checked. | |
| """ | |
| self.update() | |
| arguments_set = set(arguments) | |
| for rule in self._rules_by_endpoint[endpoint]: | |
| if arguments_set.issubset(rule.arguments): | |
| return True | |
| return False | |
| def _rules(self) -> list[Rule]: | |
| return [rule for rules in self._rules_by_endpoint.values() for rule in rules] | |
| def iter_rules(self, endpoint: t.Any | None = None) -> t.Iterator[Rule]: | |
| """Iterate over all rules or the rules of an endpoint. | |
| :param endpoint: if provided only the rules for that endpoint | |
| are returned. | |
| :return: an iterator | |
| """ | |
| self.update() | |
| if endpoint is not None: | |
| return iter(self._rules_by_endpoint[endpoint]) | |
| return iter(self._rules) | |
| def add(self, rulefactory: RuleFactory) -> None: | |
| """Add a new rule or factory to the map and bind it. Requires that the | |
| rule is not bound to another map. | |
| :param rulefactory: a :class:`Rule` or :class:`RuleFactory` | |
| """ | |
| for rule in rulefactory.get_rules(self): | |
| rule.bind(self) | |
| if not rule.build_only: | |
| self._matcher.add(rule) | |
| self._rules_by_endpoint.setdefault(rule.endpoint, []).append(rule) | |
| self._remap = True | |
| def bind( | |
| self, | |
| server_name: str, | |
| script_name: str | None = None, | |
| subdomain: str | None = None, | |
| url_scheme: str = "http", | |
| default_method: str = "GET", | |
| path_info: str | None = None, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| ) -> MapAdapter: | |
| """Return a new :class:`MapAdapter` with the details specified to the | |
| call. Note that `script_name` will default to ``'/'`` if not further | |
| specified or `None`. The `server_name` at least is a requirement | |
| because the HTTP RFC requires absolute URLs for redirects and so all | |
| redirect exceptions raised by Werkzeug will contain the full canonical | |
| URL. | |
| If no path_info is passed to :meth:`match` it will use the default path | |
| info passed to bind. While this doesn't really make sense for | |
| manual bind calls, it's useful if you bind a map to a WSGI | |
| environment which already contains the path info. | |
| `subdomain` will default to the `default_subdomain` for this map if | |
| no defined. If there is no `default_subdomain` you cannot use the | |
| subdomain feature. | |
| .. versionchanged:: 1.0 | |
| If ``url_scheme`` is ``ws`` or ``wss``, only WebSocket rules | |
| will match. | |
| .. versionchanged:: 0.15 | |
| ``path_info`` defaults to ``'/'`` if ``None``. | |
| .. versionchanged:: 0.8 | |
| ``query_args`` can be a string. | |
| .. versionchanged:: 0.7 | |
| Added ``query_args``. | |
| """ | |
| server_name = server_name.lower() | |
| if self.host_matching: | |
| if subdomain is not None: | |
| raise RuntimeError("host matching enabled and a subdomain was provided") | |
| elif subdomain is None: | |
| subdomain = self.default_subdomain | |
| if script_name is None: | |
| script_name = "/" | |
| if path_info is None: | |
| path_info = "/" | |
| # Port isn't part of IDNA, and might push a name over the 63 octet limit. | |
| server_name, port_sep, port = server_name.partition(":") | |
| try: | |
| server_name = server_name.encode("idna").decode("ascii") | |
| except UnicodeError as e: | |
| raise BadHost() from e | |
| return MapAdapter( | |
| self, | |
| f"{server_name}{port_sep}{port}", | |
| script_name, | |
| subdomain, | |
| url_scheme, | |
| path_info, | |
| default_method, | |
| query_args, | |
| ) | |
| def bind_to_environ( | |
| self, | |
| environ: WSGIEnvironment | Request, | |
| server_name: str | None = None, | |
| subdomain: str | None = None, | |
| ) -> MapAdapter: | |
| """Like :meth:`bind` but you can pass it an WSGI environment and it | |
| will fetch the information from that dictionary. Note that because of | |
| limitations in the protocol there is no way to get the current | |
| subdomain and real `server_name` from the environment. If you don't | |
| provide it, Werkzeug will use `SERVER_NAME` and `SERVER_PORT` (or | |
| `HTTP_HOST` if provided) as used `server_name` with disabled subdomain | |
| feature. | |
| If `subdomain` is `None` but an environment and a server name is | |
| provided it will calculate the current subdomain automatically. | |
| Example: `server_name` is ``'example.com'`` and the `SERVER_NAME` | |
| in the wsgi `environ` is ``'staging.dev.example.com'`` the calculated | |
| subdomain will be ``'staging.dev'``. | |
| If the object passed as environ has an environ attribute, the value of | |
| this attribute is used instead. This allows you to pass request | |
| objects. Additionally `PATH_INFO` added as a default of the | |
| :class:`MapAdapter` so that you don't have to pass the path info to | |
| the match method. | |
| .. versionchanged:: 1.0.0 | |
| If the passed server name specifies port 443, it will match | |
| if the incoming scheme is ``https`` without a port. | |
| .. versionchanged:: 1.0.0 | |
| A warning is shown when the passed server name does not | |
| match the incoming WSGI server name. | |
| .. versionchanged:: 0.8 | |
| This will no longer raise a ValueError when an unexpected server | |
| name was passed. | |
| .. versionchanged:: 0.5 | |
| previously this method accepted a bogus `calculate_subdomain` | |
| parameter that did not have any effect. It was removed because | |
| of that. | |
| :param environ: a WSGI environment. | |
| :param server_name: an optional server name hint (see above). | |
| :param subdomain: optionally the current subdomain (see above). | |
| """ | |
| env = _get_environ(environ) | |
| wsgi_server_name = get_host(env).lower() | |
| scheme = env["wsgi.url_scheme"] | |
| upgrade = any( | |
| v.strip() == "upgrade" | |
| for v in env.get("HTTP_CONNECTION", "").lower().split(",") | |
| ) | |
| if upgrade and env.get("HTTP_UPGRADE", "").lower() == "websocket": | |
| scheme = "wss" if scheme == "https" else "ws" | |
| if server_name is None: | |
| server_name = wsgi_server_name | |
| else: | |
| server_name = server_name.lower() | |
| # strip standard port to match get_host() | |
| if scheme in {"http", "ws"} and server_name.endswith(":80"): | |
| server_name = server_name[:-3] | |
| elif scheme in {"https", "wss"} and server_name.endswith(":443"): | |
| server_name = server_name[:-4] | |
| if subdomain is None and not self.host_matching: | |
| cur_server_name = wsgi_server_name.split(".") | |
| real_server_name = server_name.split(".") | |
| offset = -len(real_server_name) | |
| if cur_server_name[offset:] != real_server_name: | |
| # This can happen even with valid configs if the server was | |
| # accessed directly by IP address under some situations. | |
| # Instead of raising an exception like in Werkzeug 0.7 or | |
| # earlier we go by an invalid subdomain which will result | |
| # in a 404 error on matching. | |
| warnings.warn( | |
| f"Current server name {wsgi_server_name!r} doesn't match configured" | |
| f" server name {server_name!r}", | |
| stacklevel=2, | |
| ) | |
| subdomain = "<invalid>" | |
| else: | |
| subdomain = ".".join(filter(None, cur_server_name[:offset])) | |
| def _get_wsgi_string(name: str) -> str | None: | |
| val = env.get(name) | |
| if val is not None: | |
| return _wsgi_decoding_dance(val) | |
| return None | |
| script_name = _get_wsgi_string("SCRIPT_NAME") | |
| path_info = _get_wsgi_string("PATH_INFO") | |
| query_args = _get_wsgi_string("QUERY_STRING") | |
| return Map.bind( | |
| self, | |
| server_name, | |
| script_name, | |
| subdomain, | |
| scheme, | |
| env["REQUEST_METHOD"], | |
| path_info, | |
| query_args=query_args, | |
| ) | |
| def update(self) -> None: | |
| """Called before matching and building to keep the compiled rules | |
| in the correct order after things changed. | |
| """ | |
| if not self._remap: | |
| return | |
| with self._remap_lock: | |
| if not self._remap: | |
| return | |
| self._matcher.update() | |
| for rules in self._rules_by_endpoint.values(): | |
| rules.sort(key=lambda x: x.build_compare_key()) | |
| self._remap = False | |
| def __repr__(self) -> str: | |
| rules = self.iter_rules() | |
| return f"{type(self).__name__}({pformat(list(rules))})" | |
| class MapAdapter: | |
| """Returned by :meth:`Map.bind` or :meth:`Map.bind_to_environ` and does | |
| the URL matching and building based on runtime information. | |
| """ | |
| def __init__( | |
| self, | |
| map: Map, | |
| server_name: str, | |
| script_name: str, | |
| subdomain: str | None, | |
| url_scheme: str, | |
| path_info: str, | |
| default_method: str, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| ): | |
| self.map = map | |
| self.server_name = server_name | |
| if not script_name.endswith("/"): | |
| script_name += "/" | |
| self.script_name = script_name | |
| self.subdomain = subdomain | |
| self.url_scheme = url_scheme | |
| self.path_info = path_info | |
| self.default_method = default_method | |
| self.query_args = query_args | |
| self.websocket = self.url_scheme in {"ws", "wss"} | |
| def dispatch( | |
| self, | |
| view_func: t.Callable[[str, t.Mapping[str, t.Any]], WSGIApplication], | |
| path_info: str | None = None, | |
| method: str | None = None, | |
| catch_http_exceptions: bool = False, | |
| ) -> WSGIApplication: | |
| """Does the complete dispatching process. `view_func` is called with | |
| the endpoint and a dict with the values for the view. It should | |
| look up the view function, call it, and return a response object | |
| or WSGI application. http exceptions are not caught by default | |
| so that applications can display nicer error messages by just | |
| catching them by hand. If you want to stick with the default | |
| error messages you can pass it ``catch_http_exceptions=True`` and | |
| it will catch the http exceptions. | |
| Here a small example for the dispatch usage:: | |
| from werkzeug.wrappers import Request, Response | |
| from werkzeug.wsgi import responder | |
| from werkzeug.routing import Map, Rule | |
| def on_index(request): | |
| return Response('Hello from the index') | |
| url_map = Map([Rule('/', endpoint='index')]) | |
| views = {'index': on_index} | |
| @responder | |
| def application(environ, start_response): | |
| request = Request(environ) | |
| urls = url_map.bind_to_environ(environ) | |
| return urls.dispatch(lambda e, v: views[e](request, **v), | |
| catch_http_exceptions=True) | |
| Keep in mind that this method might return exception objects, too, so | |
| use :class:`Response.force_type` to get a response object. | |
| :param view_func: a function that is called with the endpoint as | |
| first argument and the value dict as second. Has | |
| to dispatch to the actual view function with this | |
| information. (see above) | |
| :param path_info: the path info to use for matching. Overrides the | |
| path info specified on binding. | |
| :param method: the HTTP method used for matching. Overrides the | |
| method specified on binding. | |
| :param catch_http_exceptions: set to `True` to catch any of the | |
| werkzeug :class:`HTTPException`\\s. | |
| """ | |
| try: | |
| try: | |
| endpoint, args = self.match(path_info, method) | |
| except RequestRedirect as e: | |
| return e | |
| return view_func(endpoint, args) | |
| except HTTPException as e: | |
| if catch_http_exceptions: | |
| return e | |
| raise | |
| def match( | |
| self, | |
| path_info: str | None = None, | |
| method: str | None = None, | |
| return_rule: t.Literal[False] = False, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| websocket: bool | None = None, | |
| ) -> tuple[t.Any, t.Mapping[str, t.Any]]: ... | |
| def match( | |
| self, | |
| path_info: str | None = None, | |
| method: str | None = None, | |
| return_rule: t.Literal[True] = True, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| websocket: bool | None = None, | |
| ) -> tuple[Rule, t.Mapping[str, t.Any]]: ... | |
| def match( | |
| self, | |
| path_info: str | None = None, | |
| method: str | None = None, | |
| return_rule: bool = False, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| websocket: bool | None = None, | |
| ) -> tuple[t.Any | Rule, t.Mapping[str, t.Any]]: | |
| """The usage is simple: you just pass the match method the current | |
| path info as well as the method (which defaults to `GET`). The | |
| following things can then happen: | |
| - you receive a `NotFound` exception that indicates that no URL is | |
| matching. A `NotFound` exception is also a WSGI application you | |
| can call to get a default page not found page (happens to be the | |
| same object as `werkzeug.exceptions.NotFound`) | |
| - you receive a `MethodNotAllowed` exception that indicates that there | |
| is a match for this URL but not for the current request method. | |
| This is useful for RESTful applications. | |
| - you receive a `RequestRedirect` exception with a `new_url` | |
| attribute. This exception is used to notify you about a request | |
| Werkzeug requests from your WSGI application. This is for example the | |
| case if you request ``/foo`` although the correct URL is ``/foo/`` | |
| You can use the `RequestRedirect` instance as response-like object | |
| similar to all other subclasses of `HTTPException`. | |
| - you receive a ``WebsocketMismatch`` exception if the only | |
| match is a WebSocket rule but the bind is an HTTP request, or | |
| if the match is an HTTP rule but the bind is a WebSocket | |
| request. | |
| - you get a tuple in the form ``(endpoint, arguments)`` if there is | |
| a match (unless `return_rule` is True, in which case you get a tuple | |
| in the form ``(rule, arguments)``) | |
| If the path info is not passed to the match method the default path | |
| info of the map is used (defaults to the root URL if not defined | |
| explicitly). | |
| All of the exceptions raised are subclasses of `HTTPException` so they | |
| can be used as WSGI responses. They will all render generic error or | |
| redirect pages. | |
| Here is a small example for matching: | |
| >>> m = Map([ | |
| ... Rule('/', endpoint='index'), | |
| ... Rule('/downloads/', endpoint='downloads/index'), | |
| ... Rule('/downloads/<int:id>', endpoint='downloads/show') | |
| ... ]) | |
| >>> urls = m.bind("example.com", "/") | |
| >>> urls.match("/", "GET") | |
| ('index', {}) | |
| >>> urls.match("/downloads/42") | |
| ('downloads/show', {'id': 42}) | |
| And here is what happens on redirect and missing URLs: | |
| >>> urls.match("/downloads") | |
| Traceback (most recent call last): | |
| ... | |
| RequestRedirect: http://example.com/downloads/ | |
| >>> urls.match("/missing") | |
| Traceback (most recent call last): | |
| ... | |
| NotFound: 404 Not Found | |
| :param path_info: the path info to use for matching. Overrides the | |
| path info specified on binding. | |
| :param method: the HTTP method used for matching. Overrides the | |
| method specified on binding. | |
| :param return_rule: return the rule that matched instead of just the | |
| endpoint (defaults to `False`). | |
| :param query_args: optional query arguments that are used for | |
| automatic redirects as string or dictionary. It's | |
| currently not possible to use the query arguments | |
| for URL matching. | |
| :param websocket: Match WebSocket instead of HTTP requests. A | |
| websocket request has a ``ws`` or ``wss`` | |
| :attr:`url_scheme`. This overrides that detection. | |
| .. versionadded:: 1.0 | |
| Added ``websocket``. | |
| .. versionchanged:: 0.8 | |
| ``query_args`` can be a string. | |
| .. versionadded:: 0.7 | |
| Added ``query_args``. | |
| .. versionadded:: 0.6 | |
| Added ``return_rule``. | |
| """ | |
| self.map.update() | |
| if path_info is None: | |
| path_info = self.path_info | |
| if query_args is None: | |
| query_args = self.query_args or {} | |
| method = (method or self.default_method).upper() | |
| if websocket is None: | |
| websocket = self.websocket | |
| domain_part = self.server_name | |
| if not self.map.host_matching and self.subdomain is not None: | |
| domain_part = self.subdomain | |
| path_part = f"/{path_info.lstrip('/')}" if path_info else "" | |
| try: | |
| result = self.map._matcher.match(domain_part, path_part, method, websocket) | |
| except RequestPath as e: | |
| # safe = https://url.spec.whatwg.org/#url-path-segment-string | |
| new_path = quote(e.path_info, safe="!$&'()*+,/:;=@") | |
| raise RequestRedirect( | |
| self.make_redirect_url(new_path, query_args) | |
| ) from None | |
| except RequestAliasRedirect as e: | |
| raise RequestRedirect( | |
| self.make_alias_redirect_url( | |
| f"{domain_part}|{path_part}", | |
| e.endpoint, | |
| e.matched_values, | |
| method, | |
| query_args, | |
| ) | |
| ) from None | |
| except NoMatch as e: | |
| if e.have_match_for: | |
| raise MethodNotAllowed(valid_methods=list(e.have_match_for)) from None | |
| if e.websocket_mismatch: | |
| raise WebsocketMismatch() from None | |
| raise NotFound() from None | |
| else: | |
| rule, rv = result | |
| if self.map.redirect_defaults: | |
| redirect_url = self.get_default_redirect(rule, method, rv, query_args) | |
| if redirect_url is not None: | |
| raise RequestRedirect(redirect_url) | |
| if rule.redirect_to is not None: | |
| if isinstance(rule.redirect_to, str): | |
| def _handle_match(match: t.Match[str]) -> str: | |
| value = rv[match.group(1)] | |
| return rule._converters[match.group(1)].to_url(value) | |
| redirect_url = _simple_rule_re.sub(_handle_match, rule.redirect_to) | |
| else: | |
| redirect_url = rule.redirect_to(self, **rv) | |
| if self.subdomain: | |
| netloc = f"{self.subdomain}.{self.server_name}" | |
| else: | |
| netloc = self.server_name | |
| raise RequestRedirect( | |
| urljoin( | |
| f"{self.url_scheme or 'http'}://{netloc}{self.script_name}", | |
| redirect_url, | |
| ) | |
| ) | |
| if return_rule: | |
| return rule, rv | |
| else: | |
| return rule.endpoint, rv | |
| def test(self, path_info: str | None = None, method: str | None = None) -> bool: | |
| """Test if a rule would match. Works like `match` but returns `True` | |
| if the URL matches, or `False` if it does not exist. | |
| :param path_info: the path info to use for matching. Overrides the | |
| path info specified on binding. | |
| :param method: the HTTP method used for matching. Overrides the | |
| method specified on binding. | |
| """ | |
| try: | |
| self.match(path_info, method) | |
| except RequestRedirect: | |
| pass | |
| except HTTPException: | |
| return False | |
| return True | |
| def allowed_methods(self, path_info: str | None = None) -> t.Iterable[str]: | |
| """Returns the valid methods that match for a given path. | |
| .. versionadded:: 0.7 | |
| """ | |
| try: | |
| self.match(path_info, method="--") | |
| except MethodNotAllowed as e: | |
| return e.valid_methods # type: ignore | |
| except HTTPException: | |
| pass | |
| return [] | |
| def get_host(self, domain_part: str | None) -> str: | |
| """Figures out the full host name for the given domain part. The | |
| domain part is a subdomain in case host matching is disabled or | |
| a full host name. | |
| """ | |
| if self.map.host_matching: | |
| if domain_part is None: | |
| return self.server_name | |
| return domain_part | |
| if domain_part is None: | |
| subdomain = self.subdomain | |
| else: | |
| subdomain = domain_part | |
| if subdomain: | |
| return f"{subdomain}.{self.server_name}" | |
| else: | |
| return self.server_name | |
| def get_default_redirect( | |
| self, | |
| rule: Rule, | |
| method: str, | |
| values: t.MutableMapping[str, t.Any], | |
| query_args: t.Mapping[str, t.Any] | str, | |
| ) -> str | None: | |
| """A helper that returns the URL to redirect to if it finds one. | |
| This is used for default redirecting only. | |
| :internal: | |
| """ | |
| assert self.map.redirect_defaults | |
| for r in self.map._rules_by_endpoint[rule.endpoint]: | |
| # every rule that comes after this one, including ourself | |
| # has a lower priority for the defaults. We order the ones | |
| # with the highest priority up for building. | |
| if r is rule: | |
| break | |
| if r.provides_defaults_for(rule) and r.suitable_for(values, method): | |
| values.update(r.defaults) # type: ignore | |
| domain_part, path = r.build(values) # type: ignore | |
| return self.make_redirect_url(path, query_args, domain_part=domain_part) | |
| return None | |
| def encode_query_args(self, query_args: t.Mapping[str, t.Any] | str) -> str: | |
| if not isinstance(query_args, str): | |
| return _urlencode(query_args) | |
| return query_args | |
| def make_redirect_url( | |
| self, | |
| path_info: str, | |
| query_args: t.Mapping[str, t.Any] | str | None = None, | |
| domain_part: str | None = None, | |
| ) -> str: | |
| """Creates a redirect URL. | |
| :internal: | |
| """ | |
| if query_args is None: | |
| query_args = self.query_args | |
| if query_args: | |
| query_str = self.encode_query_args(query_args) | |
| else: | |
| query_str = None | |
| scheme = self.url_scheme or "http" | |
| host = self.get_host(domain_part) | |
| path = "/".join((self.script_name.strip("/"), path_info.lstrip("/"))) | |
| return urlunsplit((scheme, host, path, query_str, None)) | |
| def make_alias_redirect_url( | |
| self, | |
| path: str, | |
| endpoint: t.Any, | |
| values: t.Mapping[str, t.Any], | |
| method: str, | |
| query_args: t.Mapping[str, t.Any] | str, | |
| ) -> str: | |
| """Internally called to make an alias redirect URL.""" | |
| url = self.build( | |
| endpoint, values, method, append_unknown=False, force_external=True | |
| ) | |
| if query_args: | |
| url += f"?{self.encode_query_args(query_args)}" | |
| assert url != path, "detected invalid alias setting. No canonical URL found" | |
| return url | |
| def _partial_build( | |
| self, | |
| endpoint: t.Any, | |
| values: t.Mapping[str, t.Any], | |
| method: str | None, | |
| append_unknown: bool, | |
| ) -> tuple[str, str, bool] | None: | |
| """Helper for :meth:`build`. Returns subdomain and path for the | |
| rule that accepts this endpoint, values and method. | |
| :internal: | |
| """ | |
| # in case the method is none, try with the default method first | |
| if method is None: | |
| rv = self._partial_build( | |
| endpoint, values, self.default_method, append_unknown | |
| ) | |
| if rv is not None: | |
| return rv | |
| # Default method did not match or a specific method is passed. | |
| # Check all for first match with matching host. If no matching | |
| # host is found, go with first result. | |
| first_match = None | |
| for rule in self.map._rules_by_endpoint.get(endpoint, ()): | |
| if rule.suitable_for(values, method): | |
| build_rv = rule.build(values, append_unknown) | |
| if build_rv is not None: | |
| rv = (build_rv[0], build_rv[1], rule.websocket) | |
| if self.map.host_matching: | |
| if rv[0] == self.server_name: | |
| return rv | |
| elif first_match is None: | |
| first_match = rv | |
| else: | |
| return rv | |
| return first_match | |
| def build( | |
| self, | |
| endpoint: t.Any, | |
| values: t.Mapping[str, t.Any] | None = None, | |
| method: str | None = None, | |
| force_external: bool = False, | |
| append_unknown: bool = True, | |
| url_scheme: str | None = None, | |
| ) -> str: | |
| """Building URLs works pretty much the other way round. Instead of | |
| `match` you call `build` and pass it the endpoint and a dict of | |
| arguments for the placeholders. | |
| The `build` function also accepts an argument called `force_external` | |
| which, if you set it to `True` will force external URLs. Per default | |
| external URLs (include the server name) will only be used if the | |
| target URL is on a different subdomain. | |
| >>> m = Map([ | |
| ... Rule('/', endpoint='index'), | |
| ... Rule('/downloads/', endpoint='downloads/index'), | |
| ... Rule('/downloads/<int:id>', endpoint='downloads/show') | |
| ... ]) | |
| >>> urls = m.bind("example.com", "/") | |
| >>> urls.build("index", {}) | |
| '/' | |
| >>> urls.build("downloads/show", {'id': 42}) | |
| '/downloads/42' | |
| >>> urls.build("downloads/show", {'id': 42}, force_external=True) | |
| 'http://example.com/downloads/42' | |
| Because URLs cannot contain non ASCII data you will always get | |
| bytes back. Non ASCII characters are urlencoded with the | |
| charset defined on the map instance. | |
| Additional values are converted to strings and appended to the URL as | |
| URL querystring parameters: | |
| >>> urls.build("index", {'q': 'My Searchstring'}) | |
| '/?q=My+Searchstring' | |
| When processing those additional values, lists are furthermore | |
| interpreted as multiple values (as per | |
| :py:class:`werkzeug.datastructures.MultiDict`): | |
| >>> urls.build("index", {'q': ['a', 'b', 'c']}) | |
| '/?q=a&q=b&q=c' | |
| Passing a ``MultiDict`` will also add multiple values: | |
| >>> urls.build("index", MultiDict((('p', 'z'), ('q', 'a'), ('q', 'b')))) | |
| '/?p=z&q=a&q=b' | |
| If a rule does not exist when building a `BuildError` exception is | |
| raised. | |
| The build method accepts an argument called `method` which allows you | |
| to specify the method you want to have an URL built for if you have | |
| different methods for the same endpoint specified. | |
| :param endpoint: the endpoint of the URL to build. | |
| :param values: the values for the URL to build. Unhandled values are | |
| appended to the URL as query parameters. | |
| :param method: the HTTP method for the rule if there are different | |
| URLs for different methods on the same endpoint. | |
| :param force_external: enforce full canonical external URLs. If the URL | |
| scheme is not provided, this will generate | |
| a protocol-relative URL. | |
| :param append_unknown: unknown parameters are appended to the generated | |
| URL as query string argument. Disable this | |
| if you want the builder to ignore those. | |
| :param url_scheme: Scheme to use in place of the bound | |
| :attr:`url_scheme`. | |
| .. versionchanged:: 2.0 | |
| Added the ``url_scheme`` parameter. | |
| .. versionadded:: 0.6 | |
| Added the ``append_unknown`` parameter. | |
| """ | |
| self.map.update() | |
| if values: | |
| if isinstance(values, MultiDict): | |
| values = { | |
| k: (v[0] if len(v) == 1 else v) | |
| for k, v in dict.items(values) | |
| if len(v) != 0 | |
| } | |
| else: # plain dict | |
| values = {k: v for k, v in values.items() if v is not None} | |
| else: | |
| values = {} | |
| rv = self._partial_build(endpoint, values, method, append_unknown) | |
| if rv is None: | |
| raise BuildError(endpoint, values, method, self) | |
| domain_part, path, websocket = rv | |
| host = self.get_host(domain_part) | |
| if url_scheme is None: | |
| url_scheme = self.url_scheme | |
| # Always build WebSocket routes with the scheme (browsers | |
| # require full URLs). If bound to a WebSocket, ensure that HTTP | |
| # routes are built with an HTTP scheme. | |
| secure = url_scheme in {"https", "wss"} | |
| if websocket: | |
| force_external = True | |
| url_scheme = "wss" if secure else "ws" | |
| elif url_scheme: | |
| url_scheme = "https" if secure else "http" | |
| # shortcut this. | |
| if not force_external and ( | |
| (self.map.host_matching and host == self.server_name) | |
| or (not self.map.host_matching and domain_part == self.subdomain) | |
| ): | |
| return f"{self.script_name.rstrip('/')}/{path.lstrip('/')}" | |
| scheme = f"{url_scheme}:" if url_scheme else "" | |
| return f"{scheme}//{host}{self.script_name[:-1]}/{path.lstrip('/')}" | |