This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.0.

Source code for kombu.utils.url

"""URL Utilities."""
# flake8: noqa


from __future__ import annotations

from collections.abc import Mapping
from functools import partial
from typing import NamedTuple
from urllib.parse import parse_qsl, quote, unquote, urlparse

try:
    import ssl
    ssl_available = True
except ImportError:  # pragma: no cover
    ssl_available = False

from ..log import get_logger

safequote = partial(quote, safe='')
logger = get_logger(__name__)

[docs]class urlparts(NamedTuple): """Named tuple representing parts of the URL.""" scheme: str hostname: str port: int username: str password: str path: str query: Mapping
[docs]def parse_url(url): # type: (str) -> Dict """Parse URL into mapping of components.""" scheme, host, port, user, password, path, query = _parse_url(url) if query: keys = [key for key in query.keys() if key.startswith('ssl_')] for key in keys: if key == 'ssl_cert_reqs': query[key] = parse_ssl_cert_reqs(query[key]) if query[key] is None: logger.warning('Defaulting to insecure SSL behaviour.') if 'ssl' not in query: query['ssl'] = {} query['ssl'][key] = query[key] del query[key] return dict(transport=scheme, hostname=host, port=port, userid=user, password=password, virtual_host=path, **query)
[docs]def url_to_parts(url): # type: (str) -> urlparts """Parse URL into :class:`urlparts` tuple of components.""" scheme = urlparse(url).scheme schemeless = url[len(scheme) + 3:] # parse with HTTP URL semantics parts = urlparse('http://' + schemeless) path = parts.path or '' path = path[1:] if path and path[0] == '/' else path return urlparts( scheme, unquote(parts.hostname or '') or None, parts.port, unquote(parts.username or '') or None, unquote(parts.password or '') or None, unquote(path or '') or None, dict(parse_qsl(parts.query)), )
_parse_url = url_to_parts
[docs]def as_url(scheme, host=None, port=None, user=None, password=None, path=None, query=None, sanitize=False, mask='**'): # type: (str, str, int, str, str, str, str, bool, str) -> str """Generate URL from component parts.""" parts = [f'{scheme}://'] if user or password: if user: parts.append(safequote(user)) if password: if sanitize: parts.extend([':', mask] if mask else [':']) else: parts.extend([':', safequote(password)]) parts.append('@') parts.append(safequote(host) if host else '') if port: parts.extend([':', port]) parts.extend(['/', path]) return ''.join(str(part) for part in parts if part)
[docs]def sanitize_url(url, mask='**'): # type: (str, str) -> str """Return copy of URL with password removed.""" return as_url(*_parse_url(url), sanitize=True, mask=mask)
[docs]def maybe_sanitize_url(url, mask='**'): # type: (Any, str) -> Any """Sanitize url, or do nothing if url undefined.""" if isinstance(url, str) and '://' in url: return sanitize_url(url, mask) return url
[docs]def parse_ssl_cert_reqs(query_value): # type: (str) -> Any """Given the query parameter for ssl_cert_reqs, return the SSL constant or None.""" if ssl_available: query_value_to_constant = { 'CERT_REQUIRED': ssl.CERT_REQUIRED, 'CERT_OPTIONAL': ssl.CERT_OPTIONAL, 'CERT_NONE': ssl.CERT_NONE, 'required': ssl.CERT_REQUIRED, 'optional': ssl.CERT_OPTIONAL, 'none': ssl.CERT_NONE, } return query_value_to_constant[query_value] else: return None