Slight refactorings

Author: vaaale

packages/fetchcraft-admin/src/fetchcraft/admin/ingestion/pipeline_factory.py

@@ -15,7 +15,7 @@
 
 from fetchcraft.connector.filesystem import FilesystemConnector
 from fetchcraft.index import BaseIndex, IndexFactory, VectorIndex
-from fetchcraft.ingestion.interfaces import QueueBackend
+from fetchcraft.ingestion.interfaces import QueueBackend, Source
 from fetchcraft.ingestion.models import IngestionJob
 from fetchcraft.ingestion.pipeline import TrackedIngestionPipeline
 from fetchcraft.ingestion.repository import JobRepository, DocumentRepository, TaskRepository
@@ -158,7 +158,12 @@ def index_id(self) -> str:
         return self._index_id
 
     @abstractmethod
-    def configure_pipeline(self, pipeline: TrackedIngestionPipeline) -> None:
+    async def create_source(self, documents_path: Path) -> Source:
+        """Create a source for the pipeline."""
+        pass
+
+    @abstractmethod
+    async def configure_pipeline(self, pipeline: TrackedIngestionPipeline) -> None:
         """
         Configure the pipeline with transformations and sinks.
         
@@ -219,12 +224,12 @@ async def create_pipeline(
         self._index_id = index_id
 
         full_source_path = self._document_root / job.source_path
-        connector = FilesystemConnector(
-            path=full_source_path,
-            filter=None
-        )
+        # connector = FilesystemConnector(
+        #     path=full_source_path,
+        #     filter=None
+        # )
 
-        directories = await connector.list_directories()
+        # directories = await connector.list_directories()
 
         # Create base pipeline with system dependencies
         pipeline = TrackedIngestionPipeline(
@@ -235,19 +240,21 @@ async def create_pipeline(
             task_repo=self.task_repo,
             num_workers=self.num_workers,
             callback_base_url=self.callback_base_url,
-            context={"directories": directories}
+            context={}
+            # context={"directories": directories}
         )
 
         # Configure source if requested (skip for recovery scenarios)
         if include_source:
-            source = ConnectorSource(
-                connector=connector,
-                document_root=self._document_root,
-            )
+            # source = ConnectorSource(
+            #     connector=connector,
+            #     document_root=self._document_root,
+            # )
+            source = await self.create_source(full_source_path)
             pipeline.source(source)
 
         # Call user's configure_pipeline to add transformations and sinks
-        self.configure_pipeline(pipeline)
+        await self.configure_pipeline(pipeline)
 
         # Clear job-specific dependencies
         self._parser_map = None

packages/fetchcraft-admin/src/fetchcraft/admin/main.py

@@ -22,8 +22,11 @@
     FetchcraftIngestionPipelineFactory,
     IngestionConfig, DefaultIndexFactory,
 )
+from fetchcraft.connector import Connector
+from fetchcraft.connector.filesystem import FilesystemConnector
 from fetchcraft.document_store import MongoDBDocumentStore
 from fetchcraft.embeddings import OpenAIEmbeddings
+from fetchcraft.ingestion import Source, ConnectorSource
 from fetchcraft.ingestion.pipeline import TrackedIngestionPipeline
 from fetchcraft.ingestion.transformations import (
     AsyncParsingTransformation,
@@ -51,6 +54,8 @@ class DefaultPipelineFactory(FetchcraftIngestionPipelineFactory):
     index_factory: DefaultIndexFactory
     chunker: HierarchicalNodeParser
     parser_map: dict[str, DocumentParser]
+    directories: list[str] = []
+    connector: Connector
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
@@ -65,13 +70,24 @@ def __init__(self, **kwargs):
     - Chunking transformation
     - Vector index sink
     """
-    
-    def configure_pipeline(self, pipeline: TrackedIngestionPipeline) -> None:
+
+    async def create_source(self, documents_path: Path) -> Source:
+        """Create a source for the pipeline."""
+        return ConnectorSource(
+            connector=self.connector,
+            document_root=self._document_root,
+        )
+
+    async def configure_pipeline(self, pipeline: TrackedIngestionPipeline) -> None:
         """Configure the pipeline with default transformations and sinks."""
+        if len(self.directories) == 0:
+            self.directories = await self.connector.list_directories()
+
         pipeline.add_transformation(AsyncParsingTransformation(parser_map=self.parser_map))
         pipeline.add_transformation(ExtractKeywords())
         pipeline.add_transformation(ChunkingTransformation(chunker=self.chunker))
         pipeline.add_sink(VectorIndexSink(index_factory=self.index_factory))
+        pipeline.context({"directories": self.directories})
 
 
 
@@ -121,6 +137,11 @@ def get_ingestion_dependencies(settings: IngestionConfig):
     # Build callback URL for docling async parsing
     callback_url = f"{settings.callback_base_url}/api/tasks/callback"
 
+    connector = FilesystemConnector(
+        path=settings.documents_path,
+        filter=None
+    )
+
     parser_map = {
         "default": TextFileParser(),
         "application/pdf": RemoteDoclingParser(
@@ -132,6 +153,7 @@ def get_ingestion_dependencies(settings: IngestionConfig):
     return {
         "index_factory": index_factory,
         "chunker": chunker,
+        "connector": connector,
         "parser_map": parser_map,
     }
 

packages/fetchcraft-core/src/fetchcraft/ingestion/pipeline.py

@@ -254,7 +254,11 @@ def __init__(
         )
 
     # ========== Builder API ==========
-    
+
+    def context(self, items: dict):
+        self._context.update(items)
+        return self
+
     def source(self, src: Source) -> "TrackedIngestionPipeline":
         """
         Set the pipeline source.

packages/fetchcraft-core/src/fetchcraft/ingestion/sinks/vector_index.py

@@ -55,7 +55,7 @@ async def write(self, record: DocumentRecord, context: Optional[dict] = None) ->
                 # Get document and chunks
                 doc = DocumentNode.model_validate(record.metadata["document"])
                 chunk_dicts = record.metadata.get("chunks", [])
-                vector_index = self.vector_index or self.index_factory.create_index(VectorIndex, index_id=self.index_id)
+                vector_index = self.vector_index or self.index_factory.create_index(VectorIndex, index_id=self.index_id, context=context)
 
                 # Add new chunks
                 if chunk_dicts: