如何更改 google.auth.transport.requests Python 库中的 _DEFAULT_TIMEOUT（120 秒）？

Question

我使用此代码，在尝试连接到服务器时偶尔会超时。现在我正在尝试找到一种方法来减少超时时间，以便 refresh() 抛出异常必须比 120 秒后更快。

def _get_access_token():
  """Retrieve a valid access token that can be used to authorize requests."""

  credentials = Credentials.from_service_account_file(os.environ['GOOGLE_APPLICATION_CREDENTIALS'],
  scopes=['https://www.googleapis.com/auth/firebase.messaging'])

  request_succeeded = False

  while True: # Try until we get a token from Google.
    try:
      credentials.refresh( google.auth.transport.requests.Request() ) # Change timeout time here?
      access_token = credentials.token 
      request_succeeded = True
    except:
      print("Exception occurred, connecting to Google again...")
    if request_succeeded:
      break

  # Write access token to file:
  f = open("access_token_firebase.txt", "w")
  f.write(access_token)
  f.close()

if __name__ == "__main__":
  start_time = time.time()
  _get_access_token()
  print("Python script finished. Took: %s seconds to complete." % round(time.time() - start_time, 2) )

Answer 1

您可以使用此链接作为参考。正如您所看到的，您应该声明超时并在请求中使用它，如下所示

 def __call__(
        self,
        url,
        method="GET",
        body=None,
        headers=None,
        timeout=_DEFAULT_TIMEOUT,
        **kwargs
    ):
        """Make an HTTP request using requests.

        Args:
            url (str): The URI to be requested.
            method (str): The HTTP method to use for the request. Defaults
                to 'GET'.
            body (bytes): The payload / body in HTTP request.
            headers (Mapping[str, str]): Request headers.
            timeout (Optional[int]): The number of seconds to wait for a
                response from the server. If not specified or if None, the
                requests default timeout will be used.
            kwargs: Additional arguments passed through to the underlying
                requests :meth:`~requests.Session.request` method.

        Returns:
            google.auth.transport.Response: The HTTP response.

        Raises:
            google.auth.exceptions.TransportError: If any exception occurred.
        """
        try:
            _LOGGER.debug("Making request: %s %s", method, url)
            response = self.session.request(
                method, url, data=body, headers=headers, timeout=timeout, **kwargs
            )
            return _Response(response)
        except requests.exceptions.RequestException as caught_exc:
            new_exc = exceptions.TransportError(caught_exc)
            six.raise_from(new_exc, caught_exc)

Answer 2

我最终选择的是

from google.auth.transport import requests as google_requests
class GoogleRequestWithTimeout(google_requests.Request):
    def __call__(self, *args, **kwargs):
        return super().__call__(*args, timeout=5, **kwargs)

所以你现在可以做

credentials.refresh(GoogleRequestWithTimeout())

请参阅此处的讨论：https://github.com/googleapis/google-auth-library-python/pull/171