Repository: deepset-ai/haystack (language: python)

Issue:
Build a CSVToDocument Component
**Is your feature request related to a problem? Please describe.**
Haystack currently supports conversion of several file formats such as `.txt`, `.pdf` and `markdown`. It would be very useful to also have a Component that converts a csv format file into a list of `Document` objects.

**Describe the solution you'd like**
I would like to implement a CSVToDocument that will load CSV files into a sequence of `Document` objects. Each row of the CSV file should be translated to one document. I think this could be the best choice since it is true that each line of a CSV file is usually a different data record. 

Each row could be converted into a key:value pair so that the Document output could be the following:

`Document(id=XXX, content: 'column1: value1\ncolumn2: value2\ncolumn3: value3', meta: {'row': 0, 'source': './example.csv'})
`

**Describe alternatives you've considered**
[Unstructured](https://docs.haystack.deepset.ai/docs/unstructuredfileconverter) is already present in Haystack so there is already a method to convert .csv files. Nevertheless, I think it is useful to have a component specifically designed for this purpose that does not go through the generation of an API key to communicate with an external service.



Discussion / hints:
Hey, @tradicio!
Just a quick clarification: Unstructured can also run locally (without API keys) using Docker, as specified [here](https://haystack.deepset.ai/integrations/unstructured-file-converter).
(I understand it is not properly documented and I'll open an issue to improve docs...)
@tradicio Thank you for your feature request and the detailed description. I understand that you are looking to implement a CSVToDocument component that converts each row of a CSV file into a separate Document object, with key-value pairs representing the column data.

However, I'd like to propose an alternative approach that could offer more flexibility and generality. The Document data class in Haystack has a dataframe property intended to hold tabular data as a pandas DataFrame. This suggests that an implementation of a CSV loader could initially load the entire CSV into this dataframe property, providing a comprehensive representation of the tabular data.

To address your specific use case of converting each CSV row into individual Document objects, we could introduce a document processor component. This component could be designed to transform the loaded document into the desired format. Specifically, it could accept a Callable[[List[Document]], List[Document]] at initialization, which would be applied in the run method to process documents as needed. This approach not only meets your requirement but also lays the foundation for expanding document preprocessing to be more multimodal-friendly.

This solution combines the best of both worlds: leveraging the existing capability to handle tabular data and providing the flexibility to process and convert documents into any desired format.

I hope this alternative is helpful and look forward to your thoughts on this approach.
Thanks @CarlosFerLo for your response! 

So far, I have not extensively used the Document property allowing pandas dataframe to be read. Personally, I have found it to be of little use in a real-world application, for example in the insertion of a CSV file within a DocumentStore and a subsequent embedding of the information contained within it.

For this reason, I thought of building an alternative solution. Given a CSV file as input, a CSVToDocument Component could build as output a list of Documents representing the list of CSV rows. In my opinion, those rows contain most of the information contained in the CSV file.

Anyhow, I completely agree that your solution could be more versatile and include more use cases.
is this related to #7784 ? 
@s-a I believe that although they are both new convertors, wither one of them handles different file types with different data encoded in them.
@tradicio I have never used this capability of documents before, and I find no direct way to give it a use right now. I believe that this new components are the first start to building support for this type of data. 

Regarding your formatting of the document content on single rows, I believe it would help by adding a prefix with a little bit of context about what the data describes. Also, ignoring fields such ass ids might also give you better performance. You could store the real row inside the `dataframe` field instead.

I encourage you to open a PR on the matter, in case you need something, just contact me or mention me on a comment.
@tradicio @CarlosFerLo I have made the draft PR https://github.com/deepset-ai/haystack/pull/8307.
Looking forward to your support and feedback on the same.

Fix patch:
diff --git a/docs/pydoc/config/converters_api.yml b/docs/pydoc/config/converters_api.yml
index e8ec88c7a2..945ae37430 100644
--- a/docs/pydoc/config/converters_api.yml
+++ b/docs/pydoc/config/converters_api.yml
@@ -13,7 +13,8 @@ loaders:
         "txt",
         "output_adapter",
         "openapi_functions",
-        "docx"
+        "docx",
+        "csv"
       ]
     ignore_when_discovered: ["__init__"]
 processors:
diff --git a/haystack/components/converters/__init__.py b/haystack/components/converters/__init__.py
index bde66e1589..681ab85c35 100644
--- a/haystack/components/converters/__init__.py
+++ b/haystack/components/converters/__init__.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from haystack.components.converters.azure import AzureOCRDocumentConverter
+from haystack.components.converters.csv import CSVToDocument
 from haystack.components.converters.docx import DOCXMetadata, DOCXToDocument
 from haystack.components.converters.html import HTMLToDocument
 from haystack.components.converters.markdown import MarkdownToDocument
@@ -27,4 +28,5 @@
     "DOCXToDocument",
     "DOCXMetadata",
     "PPTXToDocument",
+    "CSVToDocument",
 ]
diff --git a/haystack/components/converters/csv.py b/haystack/components/converters/csv.py
new file mode 100644
index 0000000000..721d8cf625
--- /dev/null
+++ b/haystack/components/converters/csv.py
@@ -0,0 +1,93 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import io
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+from haystack import Document, component, logging
+from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
+from haystack.dataclasses import ByteStream
+
+logger = logging.getLogger(__name__)
+
+
+@component
+class CSVToDocument:
+    """
+    Converts CSV files to Documents.
+
+    By default, it uses UTF-8 encoding when converting files but
+    you can also set a custom encoding.
+    It can attach metadata to the resulting documents.
+
+    ### Usage example
+
+    ```python
+    from haystack.components.converters.csv import CSVToDocument
+    converter = CSVToDocument()
+    results = converter.run(sources=["sample.csv"], meta={"date_added": datetime.now().isoformat()})
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'col1,col2\now1,row1\nrow2row2\n'
+    ```
+    """
+
+    def __init__(self, encoding: str = "utf-8"):
+        """
+        Creates a CSVToDocument component.
+
+        :param encoding:
+            The encoding of the csv files to convert.
+            If the encoding is specified in the metadata of a source ByteStream,
+            it overrides this value.
+        """
+        self.encoding = encoding
+
+    @component.output_types(documents=List[Document])
+    def run(
+        self,
+        sources: List[Union[str, Path, ByteStream]],
+        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
+    ):
+        """
+        Converts a CSV file to a Document.
+
+        :param sources:
+            List of file paths or ByteStream objects.
+        :param meta:
+            Optional metadata to attach to the documents.
+            This value can be either a list of dictionaries or a single dictionary.
+            If it's a single dictionary, its content is added to the metadata of all produced documents.
+            If it's a list, the length of the list must match the number of sources, because the two lists will
+            be zipped.
+            If `sources` contains ByteStream objects, their `meta` will be added to the output documents.
+        :returns:
+            A dictionary with the following keys:
+            - `documents`: Created documents
+        """
+        documents = []
+
+        meta_list = normalize_metadata(meta, sources_count=len(sources))
+
+        for source, metadata in zip(sources, meta_list):
+            try:
+                bytestream = get_bytestream_from_source(source)
+            except Exception as e:
+                logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e)
+                continue
+            try:
+                encoding = bytestream.meta.get("encoding", self.encoding)
+                data = io.BytesIO(bytestream.data).getvalue().decode(encoding=encoding)
+            except Exception as e:
+                logger.warning(
+                    "Could not convert file {source}. Skipping it. Error message: {error}", source=source, error=e
+                )
+                continue
+
+            merged_metadata = {**bytestream.meta, **metadata}
+            document = Document(content=data, meta=merged_metadata)
+            documents.append(document)
+
+        return {"documents": documents}
diff --git a/releasenotes/notes/add-csv-converter-5c0d52f180d498f5.yaml b/releasenotes/notes/add-csv-converter-5c0d52f180d498f5.yaml
new file mode 100644
index 0000000000..e5a01c9457
--- /dev/null
+++ b/releasenotes/notes/add-csv-converter-5c0d52f180d498f5.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Add a CSV to Document converter component. Loads the file as bytes object. Adds the loaded string as a new document that can be used for further processing by the Document Splitter.


Test patch:
diff --git a/test/components/converters/test_csv_to_document.py b/test/components/converters/test_csv_to_document.py
new file mode 100644
index 0000000000..c8f0bac0f1
--- /dev/null
+++ b/test/components/converters/test_csv_to_document.py
@@ -0,0 +1,86 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+import logging
+from unittest.mock import patch
+import pandas as pd
+from pathlib import Path
+
+import pytest
+
+from haystack.dataclasses import ByteStream
+from haystack.components.converters.csv import CSVToDocument
+
+
+@pytest.fixture
+def csv_converter():
+    return CSVToDocument()
+
+
+class TestCSVToDocument:
+    def test_init(self, csv_converter):
+        assert isinstance(csv_converter, CSVToDocument)
+
+    def test_run(self, test_files_path):
+        """
+        Test if the component runs correctly.
+        """
+        bytestream = ByteStream.from_file_path(test_files_path / "csv" / "sample_1.csv")
+        bytestream.meta["file_path"] = str(test_files_path / "csv" / "sample_1.csv")
+        bytestream.meta["key"] = "value"
+        files = [bytestream, test_files_path / "csv" / "sample_2.csv", test_files_path / "csv" / "sample_3.csv"]
+        converter = CSVToDocument()
+        output = converter.run(sources=files)
+        docs = output["documents"]
+        assert len(docs) == 3
+        assert "Name,Age\r\nJohn Doe,27\r\nJane Smith,37\r\nMike Johnson,47\r\n" == docs[0].content
+        assert isinstance(docs[0].content, str)
+        assert docs[0].meta == bytestream.meta
+        assert docs[1].meta["file_path"] == str(files[1])
+        assert docs[2].meta["file_path"] == str(files[2])
+
+    def test_run_error_handling(self, test_files_path, caplog):
+        """
+        Test if the component correctly handles errors.
+        """
+        paths = [
+            test_files_path / "csv" / "sample_2.csv",
+            "non_existing_file.csv",
+            test_files_path / "csv" / "sample_3.csv",
+        ]
+        converter = CSVToDocument()
+        with caplog.at_level(logging.WARNING):
+            output = converter.run(sources=paths)
+            assert "non_existing_file.csv" in caplog.text
+        docs = output["documents"]
+        assert len(docs) == 2
+        assert docs[0].meta["file_path"] == str(paths[0])
+
+    def test_encoding_override(self, test_files_path, caplog):
+        """
+        Test if the encoding metadata field is used properly
+        """
+        bytestream = ByteStream.from_file_path(test_files_path / "csv" / "sample_1.csv")
+        bytestream.meta["key"] = "value"
+
+        converter = CSVToDocument(encoding="utf-16-le")
+        output = converter.run(sources=[bytestream])
+        with caplog.at_level(logging.ERROR):
+            output = converter.run(sources=[bytestream])
+            assert "codec can't decode" in caplog.text
+
+        converter = CSVToDocument(encoding="utf-8")
+        output = converter.run(sources=[bytestream])
+        assert "Name,Age\r\n" in output["documents"][0].content
+
+    def test_run_with_meta(self):
+        bytestream = ByteStream(
+            data=b"Name,Age,City\r\nAlice,30,New York\r\nBob,25,Los Angeles\r\nCharlie,35,Chicago\r\n",
+            meta={"name": "test_name", "language": "en"},
+        )
+        converter = CSVToDocument()
+        output = converter.run(sources=[bytestream], meta=[{"language": "it"}])
+        document = output["documents"][0]
+
+        # check that the metadata from the bytestream is merged with that from the meta parameter
+        assert document.meta == {"name": "test_name", "language": "it"}
diff --git a/test/test_files/csv/sample_1.csv b/test/test_files/csv/sample_1.csv
new file mode 100644
index 0000000000..d3fc3116b1
--- /dev/null
+++ b/test/test_files/csv/sample_1.csv
@@ -0,0 +1,4 @@
+Name,Age
+John Doe,27
+Jane Smith,37
+Mike Johnson,47
diff --git a/test/test_files/csv/sample_2.csv b/test/test_files/csv/sample_2.csv
new file mode 100644
index 0000000000..3e23848a71
--- /dev/null
+++ b/test/test_files/csv/sample_2.csv
@@ -0,0 +1,4 @@
+Name,City
+John Doe,New York
+Jane Smith,Los Angeles
+Mike Johnson,Chicago
diff --git a/test/test_files/csv/sample_3.csv b/test/test_files/csv/sample_3.csv
new file mode 100644
index 0000000000..fcc805b0a3
--- /dev/null
+++ b/test/test_files/csv/sample_3.csv
@@ -0,0 +1,4 @@
+Name,Email
+John Doe,johndoe@example.com
+Jane Smith,janesmith@example.com
+Mike Johnson,mikejohnson@example.com