Repository: deepset-ai/haystack
Pull request: Add Crawler support for indexing pipeline 
**Proposed changes**:
- Add Crawler support for index pipeline 

**Status (please check what you already did)**:
- [X] First draft (up for discussions & feedback)
- [X] Added tests

closes #1322 

Problem statement:
Add Crawler support for Indexing Pipelines
In the current implementation, `Crawler.crawl()` writes documents to JSON files and returns their paths.

Adding an option to return documents will enable connecting a Crawler in an Indexing Pipeline where crawled documents can be indexed directly to a document store.

Hints from issue discussion:

Commit message: [crawler] Url updated.

Patch to review:
diff --git a/haystack/connector/crawler.py b/haystack/connector/crawler.py
index d1809d71a4..f8a8cf01c2 100644
--- a/haystack/connector/crawler.py
+++ b/haystack/connector/crawler.py
@@ -64,7 +64,8 @@ def crawl(self,  output_dir: Union[str, Path, None] = None,
               urls: Optional[List[str]] = None,
               crawler_depth: Optional[int] = None,
               filter_urls: Optional[List] = None,
-              overwrite_existing_files: Optional[bool] = None) -> List[Path]:
+              overwrite_existing_files: Optional[bool] = None,
+              return_documents: Optional[bool] = False) -> Union[List[Path], List[Dict]]:
         """
         Craw URL(s), extract the text from the HTML, create a Haystack Document object out of it and save it (one JSON
         file per URL, including text and basic meta data).
@@ -80,8 +81,9 @@ def crawl(self,  output_dir: Union[str, Path, None] = None,
         :param filter_urls: Optional list of regular expressions that the crawled URLs must comply with.
                            All URLs not matching at least one of the regular expressions will be dropped.
         :param overwrite_existing_files: Whether to overwrite existing files in output_dir with new content
+        :param return_documents: Return json files content
 
-        :return: List of paths where the crawled webpages got stored
+        :return: List of paths where the crawled webpages got stored or Return json files content
         """
         # use passed params or fallback to instance attributes
         urls = urls or self.urls
@@ -98,22 +100,19 @@ def crawl(self,  output_dir: Union[str, Path, None] = None,
         if not output_dir.exists():
             output_dir.mkdir(parents=True)
 
+        crawled_data: list = []
         is_not_empty = len(list(output_dir.rglob("*"))) > 0
         if is_not_empty and not overwrite_existing_files:
             logger.info(
                 f"Found data stored in `{output_dir}`. Delete this first if you really want to fetch new data."
             )
-            return []
         else:
             logger.info(f"Fetching from {urls} to `{output_dir}`")
-
-            filepaths = []
-
             sub_links: Dict[str, List] = {}
 
             # don't go beyond the initial list of urls
             if crawler_depth == 0:
-                filepaths += self._write_to_files(urls, output_dir=output_dir)
+                crawled_data += self._write_to_files(urls, output_dir=output_dir)
             # follow one level of sublinks
             elif crawler_depth == 1:
                 for url_ in urls:
@@ -121,9 +120,12 @@ def crawl(self,  output_dir: Union[str, Path, None] = None,
                     sub_links[url_] = list(self._extract_sublinks_from_url(base_url=url_, filter_urls=filter_urls,
                                                                      existed_links=existed_links))
                 for url in sub_links:
-                    filepaths += self._write_to_files(sub_links[url], output_dir=output_dir, base_url=url)
+                    crawled_data += self._write_to_files(sub_links[url], output_dir=output_dir, base_url=url)
+
+        if return_documents:
+            crawled_data = [self._read_json_file(_file) for _file in crawled_data]
 
-            return filepaths
+        return crawled_data
 
     def _write_to_files(self, urls: List[str], output_dir: Path, base_url: str = None) -> List[Path]:
         paths = []
@@ -148,9 +150,10 @@ def _write_to_files(self, urls: List[str], output_dir: Path, base_url: str = Non
 
         return paths
 
-    def run(self, output_dir: Union[str, Path, None] = None, urls: Optional[List[str]] = None, # type: ignore
-            crawler_depth: Optional[int] = None, filter_urls: Optional[List] = None, # type: ignore
-            overwrite_existing_files: Optional[bool] = None, **kwargs) -> Tuple[Dict, str]: # type: ignore
+    def run(self, output_dir: Union[str, Path, None] = None, urls: Optional[List[str]] = None,  # type: ignore
+            crawler_depth: Optional[int] = None, filter_urls: Optional[List] = None,  # type: ignore
+            overwrite_existing_files: Optional[bool] = None, return_documents: Optional[bool] = False,  # type: ignore
+            **kwargs) -> Tuple[Dict, str]:  # type: ignore
         """
         Method to be executed when the Crawler is used as a Node within a Haystack pipeline.
 
@@ -162,13 +165,18 @@ def run(self, output_dir: Union[str, Path, None] = None, urls: Optional[List[str
         :param filter_urls: Optional list of regular expressions that the crawled URLs must comply with.
                            All URLs not matching at least one of the regular expressions will be dropped.
         :param overwrite_existing_files: Whether to overwrite existing files in output_dir with new content
+        :param return_documents:  Return json files content
 
         :return: Tuple({"paths": List of filepaths, ...}, Name of output edge)
         """
 
-        filepaths = self.crawl(urls=urls, output_dir=output_dir, crawler_depth=crawler_depth, filter_urls=filter_urls,
-                               overwrite_existing_files=overwrite_existing_files)
-        results = {"paths": filepaths}
+        crawled_data = self.crawl(urls=urls, output_dir=output_dir, crawler_depth=crawler_depth,
+                                  filter_urls=filter_urls, overwrite_existing_files=overwrite_existing_files,
+                                  return_documents=return_documents)
+        if return_documents:
+            results = {"documents": crawled_data}
+        else:
+            results = {"paths": crawled_data}
         results.update(**kwargs)
         return results, "output_1"
 
@@ -207,3 +215,8 @@ def _extract_sublinks_from_url(self, base_url: str,
                         sub_links.add(sub_link)
 
         return sub_links
+
+    def _read_json_file(self, file_path: Path):
+        """Read the json file and return the content"""
+        with open(file_path.absolute(), "r") as read_file:
+            return json.load(read_file)
diff --git a/haystack/pipeline.py b/haystack/pipeline.py
index b464ca19e1..f2edbf12b1 100644
--- a/haystack/pipeline.py
+++ b/haystack/pipeline.py
@@ -34,6 +34,7 @@
 from haystack.translator.base import BaseTranslator
 from haystack.knowledge_graph.base import BaseKnowledgeGraph
 from haystack.graph_retriever.base import BaseGraphRetriever
+from haystack.connector import Crawler
 
 
 logger = logging.getLogger(__name__)
diff --git a/test/test_connector.py b/test/test_connector.py
index a9cca0028a..f733bb4e46 100644
--- a/test/test_connector.py
+++ b/test/test_connector.py
@@ -92,3 +92,21 @@ def test_crawler_content(tmp_path):
                 for partial_line in partial_content:
                     assert search(partial_line, content['text'])
                     assert partial_line in content['text']
+
+
+def test_crawler_return_document(tmp_path):
+    tmp_dir = tmp_path / "crawled_files"
+    _url = ["https://haystack.deepset.ai/docs/v0.5.0/intromd"]
+
+    crawler = Crawler(output_dir=tmp_dir)
+    docs_path = crawler.crawl(urls=_url, crawler_depth=1)
+    documents = crawler.crawl(urls=_url, crawler_depth=1, return_documents=True)
+
+    for json_file, document in zip(docs_path, documents):
+        assert isinstance(json_file, Path)
+        assert isinstance(document, dict)
+
+        with open(json_file.absolute(), "r") as read_file:
+            file_content = json.load(read_file)
+            assert file_content['meta'] == document['meta']
+            assert file_content['text'] == document['text']


Code review comments:
diff_hunk: @@ -98,32 +100,32 @@ def crawl(self,  output_dir: Union[str, Path, None] = None,
         if not output_dir.exists():
             output_dir.mkdir(parents=True)
 
+        crawled_data: list = []
         is_not_empty = len(list(output_dir.rglob("*"))) > 0
         if is_not_empty and not overwrite_existing_files:
             logger.info(
                 f"Found data stored in `{output_dir}`. Delete this first if you really want to fetch new data."
             )
-            return []
         else:
             logger.info(f"Fetching from {urls} to `{output_dir}`")
-
-            filepaths = []
-
             sub_links: Dict[str, List] = {}
 
             # don't go beyond the initial list of urls
             if crawler_depth == 0:
-                filepaths += self._write_to_files(urls, output_dir=output_dir)
+                crawled_data += self._write_to_files(urls, output_dir=output_dir)
             # follow one level of sublinks
             elif crawler_depth == 1:
                 for url_ in urls:
                     existed_links: List = list(sum(list(sub_links.values()), []))
                     sub_links[url_] = list(self._extract_sublinks_from_url(base_url=url_, filter_urls=filter_urls,
                                                                      existed_links=existed_links))
                 for url in sub_links:
-                    filepaths += self._write_to_files(sub_links[url], output_dir=output_dir, base_url=url)
+                    crawled_data += self._write_to_files(sub_links[url], output_dir=output_dir, base_url=url)
+
+        if return_documents:
path: haystack/connector/crawler.py
text: @user1:
How about moving the conversion of documents to the `run()` method? It'd simplify the `crawl()` method to always return file paths.

@author:
@user1 I did this change to provide the same interface for **run** and **crawl** method for end-user. If you think it better option then I will refactor the code.