When we use one of the google cloud python clients to communicate with GCP, there may be errors occurring sometime calling the client API methods.
For example, for the pubsub client,
a lot of the method provide retry parameter to let you configure how you want to retry.
We can use the Retry from the google api.core to customize how our retry policy will be.
from google.api_core.retry import Retry
from google.api_core import exceptions
def is_retriable_excpetion(exc) -> bool:
# In the following page, we can find a list exceptions and whether they should be retried:
# https://cloud.google.com/pubsub/docs/reference/error-codes
my_exceptions = (
exceptions.BadGateway,
exceptions.DeadlineExceeded,
)
return isinstance(exc, my_exceptions)
my_retry_policy = Retry(
predicate=is_retriable_excpetion
)The predicate parameter from Retry class accept a function.
The function receives an exception and check whether it can be retried.
You can customize a list of exceptions inside the function.
We can also use if_exception_type from api_core to create a predicate:
my_retry_policy = Retry(
predicate=api_core.retry.if_exception_type(
api_core.exceptions.Aborted,
api_core.exceptions.DeadlineExceeded,
api_core.exceptions.InternalServerError,
api_core.exceptions.ResourceExhausted,
api_core.exceptions.ServiceUnavailable,
api_core.exceptions.Unknown,
api_core.exceptions.Cancelled,
),
)The Retry class has also other parameters to configure the minimum and maximum delay between retries.
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。