如何在 python 订阅者中捕获 google.auth.exceptions.RefreshError？

Question

我正在尝试捕获以下异常，但没有任何运气。

ERROR 2019-03-04 16:31:50,522 _plugin_wrapping 15 140108141426432 AuthMetadataPluginCallback "<google.auth.transport.grpc.AuthMetadataPlugin object at 0x7f6dd8330b70>" raised exception!
Traceback (most recent call last):
File "/usr/local/lib/python3.6/site-packages/grpc/_plugin_wrapping.py", line 79, in __call__
callback_state, callback))
File "/usr/local/lib/python3.6/site-packages/google/auth/transport/grpc.py", line 77, in __call__
callback(self._get_authorization_headers(context), None)
File "/usr/local/lib/python3.6/site-packages/google/auth/transport/grpc.py", line 65, in _get_authorization_headers
  headers)
File "/usr/local/lib/python3.6/site-packages/google/auth/credentials.py", line 122, in before_request
  self.refresh(request)
File "/usr/local/lib/python3.6/site-packages/google/oauth2/service_account.py", line 322, in refresh
  request, self._token_uri, assertion)
File "/usr/local/lib/python3.6/site-packages/google/oauth2/_client.py", line 145, in jwt_grant
  response_data = _token_endpoint_request(request, token_uri, body)
File "/usr/local/lib/python3.6/site-packages/google/oauth2/_client.py", line 111, in _token_endpoint_request
  _handle_error_response(response_body)
File "/usr/local/lib/python3.6/site-packages/google/oauth2/_client.py", line 61, in _handle_error_response
  error_details, response_body)
google.auth.exceptions.RefreshError: ('invalid_grant: Invalid JWT Signature.', '{\n  "error": "invalid_grant",\n  "error_description": "Invalid JWT Signature."\n}')

我正在使用谷歌云 pubsub_v1，这部分是订阅者。 我想捕获此异常，以便客户端可以再次请求凭据。我们制作的请求凭据 API 工作正常。为了得到这个错误，我使用谷歌云控制台 UI 从 SA 中手动删除了密钥，并注释掉了启动函数调用以获取凭据。所以它正在尝试使用旧的。

在代码中，我认为我有一个非常基本的订阅者。

from google.cloud import pubsub_v1
from google.oauth2 import service_account
from google.api_core.exceptions import NotFound
from google.auth.exceptions import RefreshError

...

def startSubscribe(self):
    project, subscription, gauth = self.decryptCreds()
    credentials = service_account.Credentials.from_service_account_info(gauth)
    subscriber = pubsub_v1.SubscriberClient(credentials=credentials)
    subscription_path = subscriber.subscription_path(project, subscription)
    self.future = subscriber.subscribe(subscription_path, callback=self.callback)

    LOGGER.info('Listening for messages on {}'.format(subscription_path))
    while True:
        try:
            LOGGER.info("Calling future.result()")
            self.future.result()
            self.future.exception()
        # TODO: Need to figure out permission error, etc... and handle them properly.  This is just a catch all until
        # I figure out more about gcp subscribe and futures
        except RefreshError as e:
            # Catch permission exception and get creds again
            LOGGER.info("Caught the correct error")
            LOGGER.error(e, exc_info=True)
        except Exception as e:
            LOGGER.info("Must be something else")
            LOGGER.error(e, exc_info=True)
        LOGGER.info("Looping")

startSubscribe 函数也包含在 try/except 中，但我也没有收到错误。

有没有人有任何经验或知道如何捕获此异常，或有任何见解如何检查凭据以便我可以请求新的？

谢谢。

Answer 1

我们联系了谷歌，他们反馈说异步（使用期货）版本存在错误。

所以我修改了我们的代码以同步进行。

subscriber = None
    while self.alive:
        try:
            if subscriber is None:
                project, subscription, gauth = self.decryptCreds()
                credentials = (service_account.Credentials
                               .from_service_account_info(gauth))
                subscriber = (pubsub_v1
                              .SubscriberClient(credentials=credentials))
                subscription_path = (subscriber
                                     .subscription_path(project,
                                                        subscription))
            pull_response = subscriber.pull(subscription_path,
                                            max_messages=NUM_MESSAGES,
                                            timeout=60,
                                            retry=None)
            for msg in pull_response.received_messages:
                msg_id = msg.message.message_id

                try:
                    payload = json.loads(
                        base64.b64decode(msg.message.data.decode('utf-8'))
                        .decode('utf-8'))

                    LOGGER.info("Got Message")
                    LOGGER.debug(payload)
                    LOGGER.debug(type(payload))

                    # Removed for privacy...
                except Exception as e:
                    LOGGER.info(e, exc_info=True)
                # Here we are ack-ing the message no matter what
                # because if the message itself is bad, there is no way
                # to tell the publisher and we do not want bad messages
                # continually sent.
                subscriber.acknowledge(subscription_path, [msg.ack_id])

        except NotFound as nf:
            LOGGER.error(nf, exc_info=True)
            try:
                # If we get here try to get the credentials before remaking
                # the subscriber
                self.cpps.doGetCreds()
                # Close the channel first to make sure not file handlers
                # left open
                try:
                    subscriber.api.transport._channel.close()
                except Exception as e:
                    LOGGER.error(e, exc_info=True)
                subscriber = None
            except Exception as e:
                LOGGER.error(
                    'Exception here may have been crashing firmware.')
                LOGGER.error(e, exc_info=True)
        except (RefreshError, ServiceUnavailable) as re:
            LOGGER.error(re, exc_info=True)
        except RetryError as e:
            LOGGER.error(e, exc_info=True)
        except Exception as e:
            LOGGER.error(e, exc_info=True)

        time.sleep(10)
    LOGGER.info("Exiting Cloud Pull")
    try:
        subscriber.api.transport._channel.close()
    except Exception as e:
        LOGGER.error(e, exc_info=True)

这里需要注意的重要一点是subscriber.api.transport._channel.close()，因为它没有在文档中描述。否则会留下很多文件句柄打开。

所以基本上只要切换到同步拉取，并确保关闭通道。

希望对@ProGirlXOXO 有所帮助

哦，相关的谷歌进口是

from google.cloud import pubsub_v1
from google.oauth2 import service_account
from google.auth.exceptions import RefreshError
from google.api_core.exceptions import RetryError, ServiceUnavailable, NotFound