feat: add signed url to gcs

cloud/google/storage.py

@@ -1,5 +1,7 @@
+import datetime
 from typing import Any
 
+from google import auth
 from google.cloud import storage
 
 
@@ -15,10 +17,29 @@ def download(self, bucket_name: str, source_filename: str, destination_filename:
         blob = self.client.bucket(bucket_name).blob(source_filename)
         blob.download_to_filename(destination_filename)
 
+    def generate_presigned_url(
+        self,
+        bucket_name: str,
+        source_filename: str,
+        expiration: int,
+    ) -> str:
+        credentials, _ = auth.default()
+        client = storage.Client(credentials=credentials)
+        blob = client.bucket(bucket_name).blob(source_filename)
+
+        url = blob.generate_signed_url(
+            expiration=datetime.timedelta(seconds=expiration),
+        )
+        return url
+
 
 class Uploader(Client):
     pass
 
 
 class Downloader(Client):
     pass
+
+
+class URLSigner(Client):
+    pass

tests/google/test_storage.py

@@ -1,7 +1,8 @@
+import datetime
 import unittest
 from unittest import mock
 
-from cloud.google.storage import Downloader, Uploader
+from cloud.google.storage import Downloader, Uploader, URLSigner
 from cloud.protocols import StorageDownloader, StorageUploader
 
 
@@ -51,3 +52,28 @@ def test_storage_downloader(self, client_mock):
 
         blob = bucket.blob.return_value
         blob.download_to_filename.assert_called_once_with(destination)
+
+
+class TestStorageURLSigner(unittest.TestCase):
+    @mock.patch("google.cloud.storage.Client")
+    def test_storage_urlsigner(self, client_mock):
+        urlsigner = URLSigner()
+
+        client_mock.assert_called_once_with()
+
+        bucket_name = "test-bucket"
+        object_name = "test.txt"
+        expiration = 3600
+
+        urlsigner.generate_presigned_url(bucket_name, object_name, expiration)
+
+        cli = client_mock.return_value
+        cli.bucket.assert_called_once_with(bucket_name)
+
+        bucket = cli.bucket.return_value
+        bucket.blob.assert_called_once_with(object_name)
+
+        blob = bucket.blob.return_value
+        blob.generate_signed_url.assert_called_once_with(
+            expiration=datetime.timedelta(seconds=expiration),
+        )