feat: add Google Cloud Storage support (GCS)

openassessment/fileupload/backends/__init__.py

@@ -3,7 +3,7 @@
 
 from django.conf import settings
 
-from . import django_storage, filesystem, s3, swift
+from . import django_storage, filesystem, gcs, s3, swift
 
 
 def get_backend():
@@ -18,6 +18,8 @@ def get_backend():
         return filesystem.Backend()
     elif backend_setting == "swift":
         return swift.Backend()
+    elif backend_setting == "gcs":
+        return gcs.Backend()
     elif backend_setting == "django":
         return django_storage.Backend()
     else:

openassessment/fileupload/backends/gcs.py

@@ -0,0 +1,70 @@
+"""GCS Bucket File Upload Backend."""
+import functools
+import logging
+
+from ..exceptions import FileUploadInternalError
+from .base import BaseBackend
+
+log = logging.getLogger("openassessment.fileupload.api")  # pylint: disable=invalid-name
+
+
+def catch_broad_exception(method):
+    """Decorator to catch broad exceptions, log them, and raise a FileUploadInternalError."""
+    @functools.wraps(method)
+    def wrapper(*args, **kwargs):
+        try:
+            return method(*args, **kwargs)
+        except Exception as ex:  # pylint: disable=broad-except
+            log.exception(
+                f"Internal exception occurred while executing ora2 file-upload backend gcs.{method.__name__}: {str(ex)}"
+            )
+            raise FileUploadInternalError(ex) from ex
+    return wrapper
+
+
+class Backend(BaseBackend):
+    """ GCS Bucket File Upload Backend. """
+
+    @catch_broad_exception
+    def get_upload_url(self, key, content_type):
+        """Get a signed URL for uploading a file to GCS"""
+        bucket_name, key_name = self._retrieve_parameters(key)
+        blob = get_blob_object(bucket_name, key_name)
+        return blob.generate_signed_url(
+            version="v4",
+            expiration=self.UPLOAD_URL_TIMEOUT,
+            method="PUT",
+            content_type=content_type,
+        )
+
+    @catch_broad_exception
+    def get_download_url(self, key):
+        """Get a signed URL for downloading a file from GCS"""
+        bucket_name, key_name = self._retrieve_parameters(key)
+        blob = get_blob_object(bucket_name, key_name)
+        if not blob.exists():
+            return ""
+        return blob.generate_signed_url(
+            version="v4",
+            expiration=self.DOWNLOAD_URL_TIMEOUT,
+            method="GET",
+        )
+
+    @catch_broad_exception
+    def remove_file(self, key):
+        """Remove a file from GCS"""
+        bucket_name, key_name = self._retrieve_parameters(key)
+        blob = get_blob_object(bucket_name, key_name)
+        if blob.exists():
+            blob.delete()
+            return True
+        return False
+
+
+def get_blob_object(bucket_name, key_name):
+    """Get a blob object from GCS"""
+    # By default; avoid the need of google-cloud-storage library. It will be only needed if gcs backend is used.
+    from google.cloud import storage  # pylint: disable=import-outside-toplevel, import-error
+
+    client = storage.Client()
+    return client.bucket(bucket_name).blob(key_name)