sofima-testing/cloud_utils.py at master · jwong-nd/sofima-testing · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from google.auth import compute_engine
from google.cloud import storage

def write_to_bucket_gcs(bucket_name: str,
                        blob_name: str,
                        local_path: str,
                        project: str = 'neural-dynamics-dev'):
    """
    Uploads disk data -> cloud.

    bucket_name: public GCS bucket
    blob_name: cloud path
    local_path: data to upload
    """
    credentials = compute_engine.Credentials()
    client = storage.Client(credentials=credentials, project=project)

    bucket = client.get_bucket(bucket_name)
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(local_path)

    return blob.public_url

def read_from_bucket_gcs(bucket_name: str,
                         blob_name: str,
                         local_path: str,
                         project: str = 'neural-dynamics-dev'):
    """
    Downloads cloud data -> disk.

    bucket_name: public GCS bucket
    blob_name: cloud path to download from
    local_path: disk path
    """

    credentials = compute_engine.Credentials()
    client = storage.Client(credentials=credentials, project=project)

    bucket = client.get_bucket(bucket_name)
    blob = bucket.blob(blob_name)
    blob.download_to_filename(local_path)


# TODO:
# Implement this when it is needed
def write_to_bucket_s3():
    pass

def read_to_bucket_s3():
    pass