Skip to content

Connect

GCS

Bases: object

Helper class for uploading and download files from Google Cloud Storage

Source code in pynoddgcs/connect.py 
 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
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class GCS(object):
    """
    Helper class for uploading and download files from Google Cloud Storage
    """
    def __init__(self):
        self.client = storage.Client.create_anonymous_client()
        self.authenticated = False

    def download(self, bucket, source, destination = None):
        """
        Downloads a file from GCS, handling both public and private buckets.

        Parameters
        ----------
        bucket: str
            the bucket name
        source: str
            the file path within the bucket
        destination: str | None
            the file path to which we would like to download the file.
            default to source path within current working directory
        """
        bucket = self.client.bucket(bucket)
        blob = bucket.blob(source)
        if destination is None:
            destination = os.path.join(os.getcwd(), source)
        blob.download_to_filename(destination)

    def authenticate(self):
        """
        Get user's default credentials on this machine.

        Returns nothing, sets the values into object attributes
        """
        # Use default credentials from the gcloud environment
        try:
            credentials, project = google.auth.default()
        except google.auth.DefaultCredentialsError:
            gcloud_login()
            credentials, project = google.auth.default()

        # Use the credentials to create a storage client
        self.authenticated = True
        self.client = storage.Client(credentials=credentials)

    def check_auth(self):
        """
        Authenticates the user if not already authenticated.
        See the `authenticate` method.
        """
        if not self.authenticated:
            self.authenticate()

    def upload(self, bucket, source, destination):
        """
        Uploads a file to GCS

        Parameters
        ----------
        bucket: str
            the bucket name
        source: str
            the file path on the local machine
        destination: str
            the file path within the bucket to which we would like 
            to upload the file.
        """
        # Use default credentials from the gcloud environment
        self.check_auth()

        # Upload the file
        bucket = self.client.bucket(bucket)
        blob = bucket.blob(destination)
        blob.upload_from_filename(source)

        print(f"File {source} uploaded to {destination}.")

    def upload_string(self, bucket, contents, destination):
        """
        Uploads a string to GCS as a file

        Parameters
        ----------
        bucket: str
            the bucket name
        source: str
            the file path on the local machine
        destination: str
            the file path within the bucket to which we would like 
            to upload the file.
        """
        # Use default credentials from the gcloud environment
        self.check_auth()

        # Upload the file
        bucket = self.client.bucket(bucket)
        blob = bucket.blob(destination)
        print(contents)
        blob.upload_from_string(contents)

        print(f"Contents uploaded to {destination}.")

authenticate()

Get user's default credentials on this machine.

Returns nothing, sets the values into object attributes

Source code in pynoddgcs/connect.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def authenticate(self):
    """
    Get user's default credentials on this machine.

    Returns nothing, sets the values into object attributes
    """
    # Use default credentials from the gcloud environment
    try:
        credentials, project = google.auth.default()
    except google.auth.DefaultCredentialsError:
        gcloud_login()
        credentials, project = google.auth.default()

    # Use the credentials to create a storage client
    self.authenticated = True
    self.client = storage.Client(credentials=credentials)

check_auth()

Authenticates the user if not already authenticated. See the authenticate method.

Source code in pynoddgcs/connect.py 
58
59
60
61
62
63
64
def check_auth(self):
    """
    Authenticates the user if not already authenticated.
    See the `authenticate` method.
    """
    if not self.authenticated:
        self.authenticate()

download(bucket, source, destination=None)

Downloads a file from GCS, handling both public and private buckets.

Parameters:

Name Type Description Default
bucket

the bucket name

 required
source

the file path within the bucket

 required
destination

the file path to which we would like to download the file. default to source path within current working directory

 None
Source code in pynoddgcs/connect.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def download(self, bucket, source, destination = None):
    """
    Downloads a file from GCS, handling both public and private buckets.

    Parameters
    ----------
    bucket: str
        the bucket name
    source: str
        the file path within the bucket
    destination: str | None
        the file path to which we would like to download the file.
        default to source path within current working directory
    """
    bucket = self.client.bucket(bucket)
    blob = bucket.blob(source)
    if destination is None:
        destination = os.path.join(os.getcwd(), source)
    blob.download_to_filename(destination)

upload(bucket, source, destination)

Uploads a file to GCS

Parameters:

Name Type Description Default
bucket

the bucket name

 required
source

the file path on the local machine

 required
destination

the file path within the bucket to which we would like to upload the file.

 required
Source code in pynoddgcs/connect.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def upload(self, bucket, source, destination):
    """
    Uploads a file to GCS

    Parameters
    ----------
    bucket: str
        the bucket name
    source: str
        the file path on the local machine
    destination: str
        the file path within the bucket to which we would like 
        to upload the file.
    """
    # Use default credentials from the gcloud environment
    self.check_auth()

    # Upload the file
    bucket = self.client.bucket(bucket)
    blob = bucket.blob(destination)
    blob.upload_from_filename(source)

    print(f"File {source} uploaded to {destination}.")

upload_string(bucket, contents, destination)

Uploads a string to GCS as a file

Parameters:

Name Type Description Default
bucket

the bucket name

 required
source

the file path on the local machine

 required
destination

the file path within the bucket to which we would like to upload the file.

 required
Source code in pynoddgcs/connect.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def upload_string(self, bucket, contents, destination):
    """
    Uploads a string to GCS as a file

    Parameters
    ----------
    bucket: str
        the bucket name
    source: str
        the file path on the local machine
    destination: str
        the file path within the bucket to which we would like 
        to upload the file.
    """
    # Use default credentials from the gcloud environment
    self.check_auth()

    # Upload the file
    bucket = self.client.bucket(bucket)
    blob = bucket.blob(destination)
    print(contents)
    blob.upload_from_string(contents)

    print(f"Contents uploaded to {destination}.")

gcloud_login()

Helper method to use the gcloud API to get user credentials Requires the gcloud CLI tool to work: https://cloud.google.com/sdk/docs/install

This method should redirect the user to a browser where they can sign in to their Google Account and grant permissions to the gcloud CLI

Once the user is authenticated, we can use access tokens from the gcloud CLI to do gcloud stuff on the user's behalf.

Source code in pynoddgcs/connect.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def gcloud_login():
    """
    Helper method to use the gcloud API to get user credentials
    Requires the gcloud CLI tool to work:
    https://cloud.google.com/sdk/docs/install

    This method should redirect the user to a browser where they can sign in
    to their Google Account and grant permissions to the gcloud CLI

    Once the user is authenticated, we can use access tokens from the gcloud CLI
    to do gcloud stuff on the user's behalf.
    """
    subprocess.run(
        ["gcloud", "auth", "application-default", "login"],
        shell=True
    )

get_access_token()

Gets an access token using the gcloud CLI tool We can then use this access token to perform gcloud operations on behalf of the user.

We cache the access token for future use, as this process may make network requests which are slow.

Returns:

Name Type Description
access_token str

an access token that can be used to perform google cloud requests.
Source code in pynoddgcs/connect.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def get_access_token() -> str:
    """Gets an access token using the gcloud CLI tool
    We can then use this access token to perform gcloud operations
    on behalf of the user.

    We cache the access token for future use, as this process
    may make network requests which are slow.    

    Returns
    -------
    access_token : str
        an access token that can be used to perform google cloud requests.
    """
    if CACHED_ACCESS_TOKEN is None:
        result = subprocess.run(
            ["gcloud", "auth", "print-access-token"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            shell=True
        )
        if result.returncode != 0:
            # getting the access token failed, probably because the user
            # isn't logged in.  Log the user in, and then try again.
            gcloud_login()
            input('''
                please complete gcloud login with your browser 
                and then hit the any key to continue
            ''')
            return get_access_token()
        CACHED_ACCESS_TOKEN = result.stdout.strip()
    return CACHED_ACCESS_TOKEN