Initial version of GCS source

packages/ragbits-document-search/pyproject.toml

@@ -36,6 +36,11 @@ dependencies = [
     "ragbits"
 ]
 
+[project.optional-dependencies]
+google-cloud-storage = [
+    "gcloud-aio-storage~=9.3.0"
+]
+
 [tool.uv]
 dev-dependencies = [
     "pre-commit~=3.8.0",

packages/ragbits-document-search/src/ragbits/document_search/documents/sources.py

@@ -4,6 +4,13 @@
 
 from pydantic import BaseModel
 
+try:
+    from gcloud.aio.storage import Storage
+
+    HAS_GCLOUD_AIO = True
+except ImportError:
+    HAS_GCLOUD_AIO = False
+
 
 class Source(BaseModel, ABC):
     """
@@ -54,3 +61,44 @@ async def fetch(self) -> Path:
             The local path to the object fetched from the source.
         """
         return self.path
+
+
+class GoogleCloudStorageSource(Source):
+    """
+    An object representing a GCS file source.
+    """
+
+    source_type: Literal["google_cloud_storage_file"] = "google_cloud_storage_file"
+
+    bucket: str
+    object_name: str
+
+    path: Path
+
+    def get_id(self) -> str:
+        """
+        Get unique identifier of the object in the source.
+
+        Returns:
+            Unique identifier.
+        """
+        return f"bucket_name: {self.bucket}\nobject_name: {self.path}"
+
+    async def fetch(self) -> Path:
+        """
+        Fetch the source.
+
+        Returns:
+            Tuple containing bucket name and file path.
+
+        Raises:
+            ImportError: If the required 'gcloud' package is not installed for Google Cloud Storage source.
+        """
+
+        if not HAS_GCLOUD_AIO:
+            raise ImportError("You need to install the 'gcloud' package to use Google Cloud Storage")
+
+        async with Storage() as client:
+            await client.download_to_filename(bucket=self.bucket, object_name=self.object_name, filename=self.path)
+
+        return self.path