Spaces:
Running
Running
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('/')}" | |