Add option for anonymous credentials in GcsStorage (#2500)

This saves a lot of test time trying to get the default credentials for GCS
nuclia · Sep 27, 2024 · c68cfd6 · c68cfd6
1 parent 570fc48
commit c68cfd6
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 15 deletions.
diff --git a/nucliadb_utils/src/nucliadb_utils/storages/gcs.py b/nucliadb_utils/src/nucliadb_utils/storages/gcs.py
@@ -458,25 +458,30 @@ def __init__(
         labels: Optional[Dict[str, str]] = None,
         url: str = "https://www.googleapis.com",
         scopes: Optional[List[str]] = None,
+        anonymous: bool = False
     ):
-        if account_credentials is None:
-            self._json_credentials = None
-        elif isinstance(account_credentials, str) and account_credentials.strip() == "":
+        if anonymous:
             self._json_credentials = None
+            self._credentials = google.auth.credentials.AnonymousCredentials()
         else:
-            self._json_credentials = json.loads(base64.b64decode(account_credentials))
+            if account_credentials is None:
+                self._json_credentials = None
+            elif isinstance(account_credentials, str) and account_credentials.strip() == "":
+                self._json_credentials = None
+            else:
+                self._json_credentials = json.loads(base64.b64decode(account_credentials))
 
-        if self._json_credentials is not None:
-            self._credentials = service_account.Credentials.from_service_account_info(
-                self._json_credentials,
-                scopes=DEFAULT_SCOPES if scopes is None else scopes,
-            )
-        else:
-            try:
-                self._credentials, self._project = google.auth.default()
-            except DefaultCredentialsError:
-                logger.warning("Setting up without credentials as couldn't find workload identity")
-                self._credentials = None
+            if self._json_credentials is not None:
+                self._credentials = service_account.Credentials.from_service_account_info(
+                    self._json_credentials,
+                    scopes=DEFAULT_SCOPES if scopes is None else scopes,
+                )
+            else:
+                try:
+                    self._credentials, self._project = google.auth.default()
+                except DefaultCredentialsError:
+                    logger.warning("Setting up without credentials as couldn't find workload identity")
+                    self._credentials = None
 
         self.source = CloudFile.GCS
         self.deadletter_bucket = deadletter_bucket

diff --git a/nucliadb_utils/src/nucliadb_utils/tests/gcs.py b/nucliadb_utils/src/nucliadb_utils/tests/gcs.py
@@ -117,6 +117,7 @@ async def gcs_storage(gcs, gcs_storage_settings: dict[str, Any]):
         deadletter_bucket=extended_storage_settings.gcs_deadletter_bucket,
         indexing_bucket=extended_storage_settings.gcs_indexing_bucket,
         labels=storage_settings.gcs_bucket_labels,
+        anonymous=True,
     )
     await storage.initialize()
     yield storage