google-gax

from google.api_core.retry import Retry from google.api_core.exceptions import ServiceUnavailable import time # --- This example demonstrates google-api-core, the replacement for google-gax --- _failure_count = 0 def might_fail_api_call(): """A dummy function simulating an API call that fails intermittently.""" global _failure_count _failure_count += 1 if _failure_count < 3: print(f"Attempt {_failure_count}: API call failing with ServiceUnavailable...") raise ServiceUnavailable("Backend service is temporarily unavailable.") print(f"Attempt {_failure_count}: API call succeeded!") return "Data Retrieved!" # Define a retry policy for ServiceUnavailable errors # This replaces similar functionality found in google-gax.retry retry_policy = Retry( predicate=Retry.if_exception_type(ServiceUnavailable), initial=1.0, # Initial delay of 1 second multiplier=2.0, # Double delay each time maximum=10.0, # Max delay of 10 seconds deadline=30.0 # Total operation timeout of 30 seconds ) try: print("\n--- Calling the API with google-api-core retry policy ---") result = retry_policy(might_fail_api_call) print(f"Final Result: {result}") except Exception as e: print(f"Operation failed after retries: {e}") # For actual Google Cloud client libraries, this retry mechanism # is often automatically configured and used internally.