如何更改google.auth.transport.requests Python库中的_DEFAULT_TIMEOUT120秒？

如何解决如何更改google.auth.transport.requests Python库中的_DEFAULT_TIMEOUT120秒？

我使用此代码，在尝试连接服务器时偶尔会超时。现在，我正在尝试寻找一种减少超时时间的方法，以使 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) )

解决方法

您可以将此link用作参考。如您所见，您应该声明超时并在如下所示的请求中使用

 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,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)