Source code for kombu.utils.url
from __future__ import absolute_import
from functools import partial
try:
from urllib.parse import parse_qsl, quote, unquote, urlparse
except ImportError:
from urllib import quote, unquote # noqa
from urlparse import urlparse, parse_qsl # noqa
from . import kwdict
from kombu.five import string_t
safequote = partial(quote, safe='')
def _parse_url(url):
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 (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,
kwdict(dict(parse_qsl(parts.query))))
[docs]def parse_url(url):
scheme, host, port, user, password, path, query = _parse_url(url)
return dict(transport=scheme, hostname=host,
port=port, userid=user,
password=password, virtual_host=path, **query)
[docs]def as_url(scheme, host=None, port=None, user=None, password=None,
path=None, query=None, sanitize=False, mask='**'):
parts = ['{0}://'.format(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='**'):
return as_url(*_parse_url(url), sanitize=True, mask=mask)
[docs]def maybe_sanitize_url(url, mask='**'):
if isinstance(url, string_t) and '://' in url:
return sanitize_url(url, mask)
return url