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

Code before:
        self._rpc_retry_wait = config.getint('core', 'rpc-retry-wait', 30)
        self._rpc_log_retries = config.getboolean('core', 'rpc-log-retries', True)

        if HAS_REQUESTS:
            self._fetcher = RequestsFetcher(requests.Session())
        else:
            self._fetcher = URLLibFetcher()

    def _wait(self):
        if self._rpc_log_retries:
            logger.info("Wait for %d seconds" % self._rpc_retry_wait)
        time.sleep(self._rpc_retry_wait)

    def _fetch(self, url_suffix, body):
        full_url = _urljoin(self._url, url_suffix)
        last_exception = None
        attempt = 0
        while attempt < self._rpc_retry_attempts:
            attempt += 1
            if last_exception:
                if self._rpc_log_retries:
                    logger.info("Retrying attempt %r of %r (max)" % (attempt, self._rpc_retry_attempts))
                self._wait()  # wait for a bit and retry
            try:
                response = self._fetcher.fetch(full_url, body, self._connect_timeout)
                break
            except self._fetcher.raises as e:
                last_exception = e
                if self._rpc_log_retries:
                    logger.warning("Failed connecting to remote scheduler %r", self._url,
                                   exc_info=True)
                continue
        else:
            raise RPCError(
                "Errors (%d attempts) when connecting to remote scheduler %r" %
                (self._rpc_retry_attempts, self._url),
                last_exception
            )
        return response

    def _request(self, url, data, attempts=3, allow_null=True):
        body = {'data': json.dumps(data)}

        for _ in range(attempts):
            page = self._fetch(url, body)
            response = json.loads(page)["response"]
            if allow_null or response is not None:
                return response
        raise RPCError("Received null response from remote scheduler %r" % self._url)



Review comment (suggestion):
for some reason i'm struggling to see where it's defined that `full_uri`, `body`, and `self._connect_timeout` are being passed to `self._fetcher.fetch()` here.

Code after:
        self._rpc_retry_wait = config.getint('core', 'rpc-retry-wait', 30)
        self._rpc_log_retries = config.getboolean('core', 'rpc-log-retries', True)

        if HAS_REQUESTS:
            self._fetcher = RequestsFetcher(requests.Session())
        else:
            self._fetcher = URLLibFetcher()

    def _get_retryer(self):
        def retry_logging(retry_state):
            if self._rpc_log_retries:
                logger.warning("Failed connecting to remote scheduler %r", self._url, exc_info=True)
                logger.info("Retrying attempt %r of %r (max)" % (retry_state.attempt_number + 1, self._rpc_retry_attempts))
                logger.info("Wait for %d seconds" % self._rpc_retry_wait)

        return Retrying(wait=wait_fixed(self._rpc_retry_wait),
                        stop=stop_after_attempt(self._rpc_retry_attempts),
                        reraise=True,
                        after=retry_logging)

    def _fetch(self, url_suffix, body):
        full_url = _urljoin(self._url, url_suffix)
        scheduler_retry = self._get_retryer()

        try:
            response = scheduler_retry(self._fetcher.fetch, full_url, body, self._connect_timeout)
        except self._fetcher.raises as e:
            raise RPCError(
                "Errors (%d attempts) when connecting to remote scheduler %r" %
                (self._rpc_retry_attempts, self._url),
                e
            )
        return response

    def _request(self, url, data, attempts=3, allow_null=True):
        body = {'data': json.dumps(data)}

        for _ in range(attempts):
            page = self._fetch(url, body)
            response = json.loads(page)["response"]
            if allow_null or response is not None:
                return response
        raise RPCError("Received null response from remote scheduler %r" % self._url)


for method_name, method in RPC_METHODS.items():
    setattr(RemoteScheduler, method_name, method)