feat: document search sources string resolver

packages/ragbits-document-search/src/ragbits/document_search/_main.py

@@ -143,19 +143,30 @@ async def search(self, query: str, config: SearchConfig | None = None) -> Sequen
     @traceable
     async def ingest(
         self,
-        documents: Sequence[DocumentMeta | Document | Source],
+        documents: str | Sequence[DocumentMeta | Document | Source],
         document_processor: BaseProvider | None = None,
     ) -> None:
-        """
-        Ingest multiple documents.
+        """Ingest documents into the search index.
 
         Args:
-            documents: The documents or metadata of the documents to ingest.
+            documents: Either:
+                - A sequence of `Document`, `DocumentMetadata`, or `Source` objects
+                - A source-specific URI string (e.g., "gcs://bucket/*") to specify source location(s), for example: 
+                    - "file:///path/to/files/*.txt"
+                    - "gcs://bucket/folder/*"
+                    - "huggingface://dataset/split/row"
             document_processor: The document processor to use. If not provided, the document processor will be
                 determined based on the document metadata.
         """
+        if isinstance(documents, str):
+            from ragbits.document_search.documents.source_resolver import SourceResolver
+
+            sources = await SourceResolver.resolve(documents)
+        else:
+            sources = documents  # type: ignore[assignment]
+
         elements = await self.processing_strategy.process_documents(
-            documents, self.document_processor_router, document_processor
+            sources, self.document_processor_router, document_processor
         )
         await self._remove_entries_with_same_sources(elements)
         await self.insert_elements(elements)

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

@@ -0,0 +1,58 @@
+from collections.abc import Sequence
+from typing import ClassVar
+
+from ragbits.document_search.documents.sources import Source
+
+
+class SourceResolver:
+    """Registry for source URI protocols and their handlers.
+
+    This class provides a mechanism to register and resolve different source protocols (like 'file://', 'gcs://', etc.)
+    to their corresponding Source implementations.
+
+    Example:
+        >>> SourceResolver.register_protocol("gcs", GCSSource)
+        >>> sources = await SourceResolver.resolve("gcs://my-bucket/path/to/files/*")
+    """
+
+    _protocol_handlers: ClassVar[dict[str, type[Source]]] = {}
+
+    @classmethod
+    def register_protocol(cls, protocol: str, source_class: type[Source]) -> None:
+        """Register a source class for a specific protocol.
+
+        Args:
+            protocol: The protocol identifier (e.g., 'file', 'gcs', 's3')
+            source_class: The Source subclass that handles this protocol
+        """
+        cls._protocol_handlers[protocol] = source_class
+
+    @classmethod
+    async def resolve(cls, uri: str) -> Sequence[Source]:
+        """Resolve a URI into a sequence of Source objects.
+
+        The URI format should be: protocol://path
+        For example:
+        - file:///path/to/files/*
+        - gcs://bucket/prefix/*
+
+        Args:
+            uri: The URI to resolve
+
+        Returns:
+            A sequence of Source objects
+
+        Raises:
+            ValueError: If the URI format is invalid or the protocol is not supported
+        """
+        try:
+            protocol, path = uri.split("://", 1)
+        except ValueError as err:
+            raise ValueError(f"Invalid URI format: {uri}. Expected format: protocol://path") from err
+
+        if protocol not in cls._protocol_handlers:
+            supported = ", ".join(sorted(cls._protocol_handlers.keys()))
+            raise ValueError(f"Unsupported protocol: {protocol}. " f"Supported protocols are: {supported}")
+
+        handler_class = cls._protocol_handlers[protocol]
+        return await handler_class.from_uri(path)