Skip to content

cloud

Module with functions to interact with GCP storage service.

Functions:

Name Description
download_from_gcloud

Download a file from GCP storage.
download_from_gcloud_as_bytes

Download a file from GCP storage as bytes stream.
get_gcloud_bucket_list

List buckets available in GCP storage.
upload_to_gcloud

Upload a local file to GCP storage.
upload_to_gcloud_from_string

Upload the content of a string to GCP storage.

Attributes:

Name Type Description
log Logger

Module logger.

log module-attribute 

log: Logger = getLogger(__name__)

Module logger.

download_from_gcloud 

download_from_gcloud(
    source_blob_name: str,
    destination_file_name: str,
    bucket_name: str,
    project: str,
) -> None

Download a file from GCP storage.

Parameters:

Name Type Description Default
source_blob_name str

blob name of the source object.

 required
destination_file_name str

local filepath for the downloaded resource.

 required
bucket_name str

bucket name.

 required
project str

GCP project ID.

 required
Source code in dlunch/cloud.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def download_from_gcloud(
    source_blob_name: str,
    destination_file_name: str,
    bucket_name: str,
    project: str,
) -> None:
    """Download a file from GCP storage.

    Args:
        source_blob_name (str): blob name of the source object.
        destination_file_name (str): local filepath for the downloaded resource.
        bucket_name (str): bucket name.
        project (str): GCP project ID.
    """
    # Create storage client
    storage_client = storage.Client(project=project)

    try:
        # Get bucket
        bucket = storage_client.bucket(bucket_name)
        # Create blob
        blob = bucket.blob(source_blob_name)
        # Download
        blob.download_to_filename(destination_file_name)
        log.info(
            f"file '{source_blob_name}' downloaded to file '{destination_file_name}' successfully"
        )
    except Exception as e:
        log.warning("google storage download exception\n\t" + str(e))

download_from_gcloud_as_bytes 

download_from_gcloud_as_bytes(
    source_blob_name: str, bucket_name: str, project: str
) -> bytes

Download a file from GCP storage as bytes stream.

Parameters:

Name Type Description Default
source_blob_name str

blob name of the source file.

 required
bucket_name str

bucket name.

 required
project str

GCP project ID.

 required

Returns:

Type Description
bytes

downloaded resource.
Source code in dlunch/cloud.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def download_from_gcloud_as_bytes(
    source_blob_name: str,
    bucket_name: str,
    project: str,
) -> bytes:
    """Download a file from GCP storage as bytes stream.

    Args:
        source_blob_name (str): blob name of the source file.
        bucket_name (str): bucket name.
        project (str): GCP project ID.

    Returns:
        bytes: downloaded resource.
    """
    # Create storage client
    storage_client = storage.Client(project=project)

    try:
        # Get bucket
        bucket = storage_client.bucket(bucket_name)
        # Create blob
        blob = bucket.blob(source_blob_name)
        # Download
        bytes_object = blob.download_as_bytes()
        log.info(
            f"file '{source_blob_name}' downloaded to object successfully"
        )
    except Exception as e:
        log.warning("google storage download exception\n\t" + str(e))

    return bytes_object

get_gcloud_bucket_list 

get_gcloud_bucket_list(project: str) -> list[str]

List buckets available in GCP storage.

Parameters:

Name Type Description Default
project str

GCP project ID.

 required

Returns:

Type Description
list[str]

list with bucket names.
Source code in dlunch/cloud.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_gcloud_bucket_list(project: str) -> list[str]:
    """List buckets available in GCP storage.

    Args:
        project (str): GCP project ID.

    Returns:
        list[str]: list with bucket names.
    """
    # Create storage client
    storage_client = storage.Client(project=project)

    # Return bucket
    buckets = list(storage_client.list_buckets())

    return buckets

upload_to_gcloud 

upload_to_gcloud(
    source_file_name: str,
    destination_blob_name: str,
    bucket_name: str,
    project: str,
) -> None

Upload a local file to GCP storage.

Parameters:

Name Type Description Default
source_file_name str

filepath.

 required
destination_blob_name str

blob name to use as destination.

 required
bucket_name str

bucket name.

 required
project str

GCP project ID.

 required
Source code in dlunch/cloud.py 
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
def upload_to_gcloud(
    source_file_name: str,
    destination_blob_name: str,
    bucket_name: str,
    project: str,
) -> None:
    """Upload a local file to GCP storage.

    Args:
        source_file_name (str): filepath.
        destination_blob_name (str): blob name to use as destination.
        bucket_name (str): bucket name.
        project (str): GCP project ID.
    """
    # Create storage client
    storage_client = storage.Client(project=project)

    try:
        # Get bucket
        bucket = storage_client.bucket(bucket_name)
        # Create blob
        blob = bucket.blob(destination_blob_name)
        # Upload
        blob.upload_from_filename(source_file_name)
        log.info(
            f"file '{source_file_name}' uploaded to bucket '{bucket_name}' successfully"
        )
    except Exception as e:
        log.warning("google storage upload exception\n\t" + str(e))

upload_to_gcloud_from_string 

upload_to_gcloud_from_string(
    source_string: str,
    destination_blob_name: str,
    bucket_name: str,
    project: str,
) -> None

Upload the content of a string to GCP storage.

Parameters:

Name Type Description Default
source_string str

string to upload.

 required
destination_blob_name str

blob name to use as destination.

 required
bucket_name str

bucket name.

 required
project str

GCP project ID.

 required
Source code in dlunch/cloud.py 
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
def upload_to_gcloud_from_string(
    source_string: str,
    destination_blob_name: str,
    bucket_name: str,
    project: str,
) -> None:
    """Upload the content of a string to GCP storage.

    Args:
        source_string (str): string to upload.
        destination_blob_name (str): blob name to use as destination.
        bucket_name (str): bucket name.
        project (str): GCP project ID.
    """
    # Create storage client
    storage_client = storage.Client(project=project)

    try:
        # Get bucket
        bucket = storage_client.bucket(bucket_name)
        # Create blob
        blob = bucket.blob(destination_blob_name)
        # Upload
        blob.upload_from_string(source_string)
        log.info(
            f"file uploaded from string to bucket '{bucket_name}' at '{destination_blob_name}' successfully"
        )
    except Exception as e:
        log.warning("google storage upload exception\n\t" + str(e))