Spaces:
Sleeping
Sleeping
| 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() | |