Spaces:
Sleeping
Sleeping
File size: 1,846 Bytes
8630bef 687f595 8630bef 4029b73 687f595 4029b73 |
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 51 52 53 |
from google.cloud import storage
from google.oauth2 import service_account
from .base_storage import BaseStorage
class GoogleCloudStorage(BaseStorage):
def __init__(self, service_account_path: str | None = None):
if service_account_path:
credentials = service_account.Credentials.from_service_account_file(
service_account_path
)
self.client = storage.Client(credentials=credentials)
else:
self.client = storage.Client.create_anonymous_client()
def upload_file_from_content(
self, bucket_name: str, bucket_file_path: str, file_content: bytes
) -> str:
"""Uploads a file to a Google Cloud Storage bucket
Args:
bucket_name (str): Bucket name
bucket_file_path (str): Path to the file to be uploaded
file_content (bytes): Path to the file to be uploaded
Returns:
str: URL of the uploaded file
"""
print(f"gs://{bucket_name}/{bucket_file_path}")
bucket = self.client.get_bucket(bucket_name)
blob = bucket.blob(f"{bucket_file_path}")
blob.upload_from_string(file_content)
print(f"File uploaded to gs://{bucket_name}/{bucket_file_path}")
return f"gs://{bucket_name}/{bucket_file_path}"
def download_blob(self, bucket_name: str, source_blob_name: str) -> bytes:
"""Downloads a blob from the bucket
Args:
bucket_name (str): Bucket name
source_blob_name (str): Name of the blob to download
Returns:
bytes: Content of the blob
"""
bucket = self.client.bucket(bucket_name)
source_blob_name = source_blob_name.replace(f"gs://{bucket_name}/", "")
blob = bucket.blob(source_blob_name)
return blob.download_as_bytes()
|