Using Tenacity for retries

google_cloud_helper.py

""" gcloud storage helper functions """
import json
import logging
import posixpath
from typing import Tuple, Any, Dict, List

from tenacity import retry, stop_after_attempt, wait_exponential
from google.cloud import storage

storage_client = storage.Client()

def my_after(retry_state):
    logging.warning('Retrying %s: attempt %s ended with: %s', retry_state.fn, retry_state.attempt_number, retry_state.outcome)
    logging.warning('Resetting client after failed call.')
    global storage_client # pylint: disable=global-statement
    storage_client = storage.Client()

@retry(stop=(stop_after_attempt(5)), wait=wait_exponential(multiplier=1, min=4, max=10), after=my_after)
def upload_to_bucket(bucket_name: str, blob_name: str, file_json: str) -> str:
    """ Upload data to a bucket"""
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(blob_name)
    blob.upload_from_string(file_json, content_type='application/json')
    return blob.public_url