Repository: spotify/luigi
File: luigi/rpc.py (Python)

Code before:
    Join relative URLs to base URLs like urllib.parse.urljoin but support
    arbitrary URIs (esp. 'http+unix://').
    """
    parsed = urlparse(base)
    scheme = parsed.scheme
    return urlparse(
        urljoin(parsed._replace(scheme='http').geturl(), url)
    )._replace(scheme=scheme).geturl()


class RPCError(Exception):

    def __init__(self, message, sub_exception=None):
        super(RPCError, self).__init__(message)
        self.sub_exception = sub_exception


class URLLibFetcher(object):
    raises = (URLError, socket.timeout)

    def _create_request(self, full_url, body=None):
        # when full_url contains basic auth info, extract it and set the Authorization header
        url = urlparse(full_url)
        if url.username:
            # base64 encoding of username:password
            auth = base64.b64encode(f'{url.username}:{url.password or ""}'.encode("latin-1"))
            if six.PY3:
                auth = auth.decode('utf-8')

            # update full_url and create a request object with the auth header set
            full_url = url._replace(netloc=url.netloc.split('@', 1)[-1]).geturl()
            req = Request(full_url)
            req.add_header('Authorization', 'Basic {}'.format(auth))
        else:
            req = Request(full_url)

        # add the request body
        if body:
            req.data = urlencode(body).encode('utf-8')

        return req

    def fetch(self, full_url, body, timeout):
        req = self._create_request(full_url, body=body)
        return urlopen(req, timeout=timeout).read().decode('utf-8')


class RequestsFetcher(object):
    def __init__(self, session):
        from requests import exceptions as requests_exceptions
        self.raises = requests_exceptions.RequestException

Review comment (bug):
I think `utf-8` is more suitable according to https://stackoverflow.com/questions/7242316/what-encoding-should-i-use-for-http-basic-authentication . `utf-8` maps ASCII bytes to ASCII bytes, so it should be backward compatible.

Code after:
    arbitrary URIs (esp. 'http+unix://').
    """
    parsed = urlparse(base)
    scheme = parsed.scheme
    return urlparse(
        urljoin(parsed._replace(scheme='http').geturl(), url)
    )._replace(scheme=scheme).geturl()


class RPCError(Exception):

    def __init__(self, message, sub_exception=None):
        super(RPCError, self).__init__(message)
        self.sub_exception = sub_exception


class URLLibFetcher(object):
    raises = (URLError, socket.timeout)

    def _create_request(self, full_url, body=None):
        # when full_url contains basic auth info, extract it and set the Authorization header
        url = urlparse(full_url)
        if url.username:
            # base64 encoding of username:password
            auth = base64.b64encode('{}:{}'.format(url.username, url.password or '').encode('utf-8'))
            auth = auth.decode('utf-8')
            # update full_url and create a request object with the auth header set
            full_url = url._replace(netloc=url.netloc.split('@', 1)[-1]).geturl()
            req = Request(full_url)
            req.add_header('Authorization', 'Basic {}'.format(auth))
        else:
            req = Request(full_url)

        # add the request body
        if body:
            req.data = urlencode(body).encode('utf-8')

        return req

    def fetch(self, full_url, body, timeout):
        req = self._create_request(full_url, body=body)
        return urlopen(req, timeout=timeout).read().decode('utf-8')


class RequestsFetcher(object):
    def __init__(self, session):
        from requests import exceptions as requests_exceptions
        self.raises = requests_exceptions.RequestException
        self.session = session
        self.process_id = os.getpid()