Skip to content

Instantly share code, notes, and snippets.

@pbrumblay

pbrumblay/google_cloud_helper.py

Last active November 10, 2020 18:50
Show Gist options
  • Save pbrumblay/11c010fc7cfbf41f0c7785cc950b2608 to your computer and use it in GitHub Desktop.
Save pbrumblay/11c010fc7cfbf41f0c7785cc950b2608 to your computer and use it in GitHub Desktop.
Download ZIP
Using Tenacity for retries
Raw
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment