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

Issue:
JSON Convertor :  Pipeline Component
**Is your feature request related to a problem? Please describe.**
Currently we have a `.txt` to `Document` convertor besides others and unstructured. But I see most of the data we deal with are in the form of JSON.

**Describe the solution you'd like**
So a `.json` to `Document` convertor will be a bread winner while consuming API data in pipelines.

**Describe alternatives you've considered**
[Unstructured file convertor](https://docs.haystack.deepset.ai/docs/unstructuredfileconverter) is present but JSON schema as a individual convertor adds more sense and value.



Discussion / hints:
@srini047 Thank you for your suggestion. If you like, feel free to open a pull request. Our contributing guidelines are [here](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md).
@julian-risch 
I like to start contributing Haystack. Do you think this can be a good starter? In that case, please feel free to assign it to me. 

Thanks, 
@arminnajafi I have been contributing for a month or so, and they do not normally assign external people to issues, or at least that I have seen. If you want to do this one, feel free to open a PR, they will review it when you are ready. 
I'm proposing a `JSONToDocument` converter for Haystack 2.0, inspired by LangChain's [JSONLoader](https://python.langchain.com/v0.1/docs/modules/data_connection/document_loaders/json/). This would allow powerful parsing of JSON files, similar to LangChain's implementation which is built on [jq](https://jqlang.github.io/jq/manual/).

Example JSON schema ([prize.json](https://api.nobelprize.org/v1/prize.json)):
```json
{
    "prizes": [
        {
            "year": "string",
            "category": "string",
            "laureates": [
                {
                    "id": "string",
                    "firstname": "string",
                    "surname": "string",
                    "motivation": "string",
                    "share": "string",
                }
            ],
        }
    ]
}
```
proposed implementation
```python
from haystack.components.converters import JSONToDocument

converter = JSONToDocument(
    jq_schema=".prizes[].laureates[]?",
    content_key="motivation",
    additional_meta_fields=["firstname", "surname", "share"],
)
docs = converter.run(sources=["./prize.json"])
print(docs["documents"][0])
```

expected output:
```
Document(id=db72dfbe9, content: '"for the discovery and synthesis of quantum dots"', meta: {'file_path': './prize.json', 'firstname': 'Moungi', 'surname': 'Bawendi', 'share': '3'})
```
Based on the @kanenorman suggestion, I realized a basic implementation of a JSONToDocument component in #8079.

In this first implementation, I have not yet included the `content_key` and `additional_meta_fields` parameters yet. The point is that I have not figured out a simple way to implement the metadata logic in any JSON structure.

Let me know how this component can be improved to include this logic.
@tradicio - Thank you. I'm working on incorporating the jq logic. Are you planning on leaving your PR up as final or converting to draft?
> @tradicio - Thank you. I'm working on incorporating the jq logic. Are you planning on leaving your PR up as final or converting to draft?

I'm not sure how much I'll be able to work on the PR in the next few weeks, if you think you can incorporate the jq logic I'm more than happy to make the PR become draft
this component would be game changer. i try  hard to find a solution for old but gold data tables :) like csv etc. @tradicio could you point me to the right direction where i can find more information about this topic? 
for any of these kind of things i could convert csv2json, xls2json(+worksheets) then?
> this component would be game changer. i try hard to find a solution for old but gold data tables :) like csv etc. @tradicio could you point me to the right direction where i can find more information about this topic? for any of these kind of things i could convert csv2json, xls2json(+worksheets) then?

For more information on how the logic behind jq works, I recommend you start with the [official documentation](https://jqlang.github.io/jq/). Regarding the structure of the component, I have been inspired by the JSONLoader component in LangChain, as suggested by @kanenorman.

With respect to your second question, I think this JSONToDocument component can be also a first step to work on tabular data but I would still keep separate any future components (such as the one I suggested in #8036). CSV and XLSX files are often used to collect data with specific structures compared to JSON files. It seems to me that they are used in different contexts and for different purposes so they need different processor components.

Fix patch:
diff --git a/docs/pydoc/config/converters_api.yml b/docs/pydoc/config/converters_api.yml
index 945ae37430..6c89138c98 100644
--- a/docs/pydoc/config/converters_api.yml
+++ b/docs/pydoc/config/converters_api.yml
@@ -4,17 +4,18 @@ loaders:
     modules:
       [
         "azure",
+        "csv",
+        "docx",
         "html",
+        "json",
         "markdown",
+        "openapi_functions",
+        "output_adapter",
         "pdfminer",
-        "pypdf",
         "pptx",
+        "pypdf",
         "tika",
         "txt",
-        "output_adapter",
-        "openapi_functions",
-        "docx",
-        "csv"
       ]
     ignore_when_discovered: ["__init__"]
 processors:
diff --git a/haystack/components/converters/__init__.py b/haystack/components/converters/__init__.py
index 681ab85c35..4561dd1e0a 100644
--- a/haystack/components/converters/__init__.py
+++ b/haystack/components/converters/__init__.py
@@ -6,6 +6,7 @@
 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.json import JSONConverter
 from haystack.components.converters.markdown import MarkdownToDocument
 from haystack.components.converters.openapi_functions import OpenAPIServiceToFunctions
 from haystack.components.converters.output_adapter import OutputAdapter
@@ -29,4 +30,5 @@
     "DOCXMetadata",
     "PPTXToDocument",
     "CSVToDocument",
+    "JSONConverter",
 ]
diff --git a/haystack/components/converters/json.py b/haystack/components/converters/json.py
new file mode 100644
index 0000000000..966da881b8
--- /dev/null
+++ b/haystack/components/converters/json.py
@@ -0,0 +1,277 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from pathlib import Path
+from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union
+
+from haystack import component, default_from_dict, default_to_dict, logging
+from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
+from haystack.dataclasses import ByteStream, Document
+from haystack.lazy_imports import LazyImport
+
+logger = logging.getLogger(__name__)
+
+with LazyImport("Run 'pip install jq'") as jq_import:
+    import jq
+
+
+@component
+class JSONConverter:
+    """
+    Converts one or more JSON files into a text document.
+
+    ### Usage examples
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    source = ByteStream.from_string(json.dumps({"text": "This is the content of my document"}))
+
+    converter = JSONConverter(content_key="text")
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'This is the content of my document'
+    ```
+
+    Optionally, you can also provide a `jq_schema` string to filter the JSON source files and `extra_meta_fields`
+    to extract from the filtered data:
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    data = {
+        "laureates": [
+            {
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced "
+                "by neutron irradiation, and for his related discovery of nuclear reactions brought about by"
+                " slow neutrons",
+            },
+            {
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+            },
+        ],
+    }
+    source = ByteStream.from_string(json.dumps(data))
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'for his demonstrations of the existence of new radioactive elements produced by
+    # neutron irradiation, and for his related discovery of nuclear reactions brought
+    # about by slow neutrons'
+
+    print(documents[0].meta)
+    # {'firstname': 'Enrico', 'surname': 'Fermi'}
+
+    print(documents[1].content)
+    # 'for their discoveries of growth factors'
+
+    print(documents[1].meta)
+    # {'firstname': 'Rita', 'surname': 'Levi-Montalcini'}
+    ```
+
+    """
+
+    def __init__(
+        self,
+        jq_schema: Optional[str] = None,
+        content_key: Optional[str] = None,
+        extra_meta_fields: Optional[Union[Set[str], Literal["*"]]] = None,
+    ):
+        """
+        Creates a JSONConverter component.
+
+        An optional `jq_schema` can be provided to extract nested data in the JSON source files.
+        See the [official jq documentation](https://jqlang.github.io/jq/) for more info on the filters syntax.
+        If `jq_schema` is not set, whole JSON source files will be used to extract content.
+
+        Optionally, you can provide a `content_key` to specify which key in the extracted object must
+        be set as the document's content.
+
+        If both `jq_schema` and `content_key` are set, the component will search for the `content_key` in
+        the JSON object extracted by `jq_schema`. If the extracted data is not a JSON object, it will be skipped.
+
+        If only `jq_schema` is set, the extracted data must be a scalar value. If it's a JSON object or array,
+        it will be skipped.
+
+        If only `content_key` is set, the source JSON file must be a JSON object, else it will be skipped.
+
+        `extra_meta_fields` can either be set to a set of strings or a literal `"*"` string.
+        If it's a set of strings, it must specify fields in the extracted objects that must be set in
+        the extracted documents. If a field is not found, the meta value will be `None`.
+        If set to `"*"`, all fields that are not `content_key` found in the filtered JSON object will
+        be saved as metadata.
+
+        Initialization will fail if neither `jq_schema` nor `content_key` are set.
+
+        :param jq_schema:
+            Optional jq filter string to extract content.
+            If not specified, whole JSON object will be used to extract information.
+        :param content_key:
+            Optional key to extract document content.
+            If `jq_schema` is specified, the `content_key` will be extracted from that object.
+        :param extra_meta_fields:
+            An optional set of meta keys to extract from the content.
+            If `jq_schema` is specified, all keys will be extracted from that object.
+        """
+        self._compiled_filter = None
+        if jq_schema:
+            jq_import.check()
+            self._compiled_filter = jq.compile(jq_schema)
+
+        self._jq_schema = jq_schema
+        self._content_key = content_key
+        self._meta_fields = extra_meta_fields
+
+        if self._compiled_filter is None and self._content_key is None:
+            msg = "No `jq_schema` nor `content_key` specified. Set either or both to extract data."
+            raise ValueError(msg)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Serializes the component to a dictionary.
+
+        :returns:
+            Dictionary with serialized data.
+        """
+        return default_to_dict(
+            self, jq_schema=self._jq_schema, content_key=self._content_key, extra_meta_fields=self._meta_fields
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "JSONConverter":
+        """
+        Deserializes the component from a dictionary.
+
+        :param data:
+            Dictionary to deserialize from.
+        :returns:
+            Deserialized component.
+        """
+        return default_from_dict(cls, data)
+
+    def _get_content_and_meta(self, source: ByteStream) -> List[Tuple[str, Dict[str, Any]]]:
+        """
+        Utility function to extract text and metadata from a JSON file.
+
+        :param source:
+            UTF-8 byte stream.
+        :returns:
+            Collection of text and metadata dict tuples, each corresponding
+            to a different document.
+        """
+        try:
+            file_content = source.data.decode("utf-8")
+        except UnicodeError as exc:
+            logger.warning(
+                "Failed to extract text from {source}. Skipping it. Error: {error}",
+                source=source.meta["file_path"],
+                error=exc,
+            )
+
+        meta_fields = self._meta_fields or set()
+
+        if self._compiled_filter is not None:
+            try:
+                objects = list(self._compiled_filter.input_text(file_content))
+            except Exception as exc:
+                logger.warning(
+                    "Failed to extract text from {source}. Skipping it. Error: {error}",
+                    source=source.meta["file_path"],
+                    error=exc,
+                )
+                return []
+        else:
+            # We just load the whole file as JSON if the user didn't provide a jq filter.
+            # We put it in a list even if it's not to ease handling it later on.
+            objects = [json.loads(file_content)]
+
+        result = []
+        if self._content_key is not None:
+            for obj in objects:
+                if not isinstance(obj, dict):
+                    logger.warning("Expected a dictionary but got {obj}. Skipping it.", obj=obj)
+                    continue
+                if self._content_key not in obj:
+                    logger.warning(
+                        "'{content_key}' not found in {obj}. Skipping it.", content_key=self._content_key, obj=obj
+                    )
+                    continue
+
+                text = obj[self._content_key]
+                if isinstance(text, (dict, list)):
+                    logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
+                    continue
+
+                meta = {}
+                if meta_fields == "*":
+                    meta = {k: v for k, v in obj.items() if k != self._content_key}
+                else:
+                    for field in meta_fields:
+                        meta[field] = obj.get(field, None)
+                result.append((text, meta))
+        else:
+            for obj in objects:
+                if isinstance(obj, (dict, list)):
+                    logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
+                    continue
+                result.append((str(obj), {}))
+
+        return result
+
+    @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 list of JSON files to documents.
+
+        :param sources:
+            A 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.
+            If `sources` contain ByteStream objects, their `meta` will be added to the output documents.
+
+        :returns:
+            A dictionary with the following keys:
+            - `documents`: A list of created documents.
+        """
+        documents = []
+        meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
+
+        for source, metadata in zip(sources, meta_list):
+            try:
+                bytestream = get_bytestream_from_source(source)
+            except Exception as exc:
+                logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=exc)
+                continue
+
+            data = self._get_content_and_meta(bytestream)
+
+            for text, extra_meta in data:
+                merged_metadata = {**bytestream.meta, **metadata, **extra_meta}
+                document = Document(content=text, meta=merged_metadata)
+                documents.append(document)
+
+        return {"documents": documents}
diff --git a/pyproject.toml b/pyproject.toml
index 5c75292f7d..be286052be 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -102,8 +102,9 @@ extra-dependencies = [
   "trafilatura",                      # HTMLToDocument
   "python-pptx",                      # PPTXToDocument
   "python-docx",                      # DocxToDocument
+  "jq",                               #JSONConverter
 
-  "nltk",                             # NLTKDocumentSplitter
+  "nltk", # NLTKDocumentSplitter
 
   # OpenAPI
   "jsonref",  # OpenAPIServiceConnector, OpenAPIServiceToFunctions
diff --git a/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml b/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml
new file mode 100644
index 0000000000..f8a269d2c0
--- /dev/null
+++ b/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml
@@ -0,0 +1,48 @@
+---
+features:
+  - |
+    Add new `JSONConverter` Component to convert JSON files to `Document`.
+    Optionally it can use jq to filter the source JSON files and extract only specific parts.
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    data = {
+        "laureates": [
+            {
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced "
+                "by neutron irradiation, and for his related discovery of nuclear reactions brought about by slow neutrons",
+            },
+            {
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+            },
+        ],
+    }
+    source = ByteStream.from_string(json.dumps(data))
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields=["firstname", "surname"]
+    )
+
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'for his demonstrations of the existence of new radioactive elements produced by
+    # neutron irradiation, and for his related discovery of nuclear reactions brought
+    # about by slow neutrons'
+
+    print(documents[0].meta)
+    # {'firstname': 'Enrico', 'surname': 'Fermi'}
+
+    print(documents[1].content)
+    # 'for their discoveries of growth factors'
+
+    print(documents[1].meta)
+    # {'firstname': 'Rita', 'surname': 'Levi-Montalcini'}
+    ```


Test patch:
diff --git a/test/components/converters/test_json.py b/test/components/converters/test_json.py
new file mode 100644
index 0000000000..fe43044f74
--- /dev/null
+++ b/test/components/converters/test_json.py
@@ -0,0 +1,433 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from unittest.mock import patch
+from pathlib import Path
+import logging
+
+import pytest
+
+from haystack.components.converters import JSONConverter
+from haystack.dataclasses import ByteStream
+
+
+test_data = [
+    {
+        "year": "1997",
+        "category": "literature",
+        "laureates": [
+            {
+                "id": "674",
+                "firstname": "Dario",
+                "surname": "Fo",
+                "motivation": "who emulates the jesters of the Middle Ages in scourging authority and upholding the dignity of the downtrodden",
+                "share": "1",
+            }
+        ],
+    },
+    {
+        "year": "1986",
+        "category": "medicine",
+        "laureates": [
+            {
+                "id": "434",
+                "firstname": "Stanley",
+                "surname": "Cohen",
+                "motivation": "for their discoveries of growth factors",
+                "share": "2",
+            },
+            {
+                "id": "435",
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+                "share": "2",
+            },
+        ],
+    },
+    {
+        "year": "1938",
+        "category": "physics",
+        "laureates": [
+            {
+                "id": "46",
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced by neutron irradiation, and for his related discovery of nuclear reactions brought about by slow neutrons",
+                "share": "1",
+            }
+        ],
+    },
+]
+
+
+def test_init_without_jq_schema_and_content_key():
+    with pytest.raises(
+        ValueError, match="No `jq_schema` nor `content_key` specified. Set either or both to extract data."
+    ):
+        JSONConverter()
+
+
+@patch("haystack.components.converters.json.jq_import")
+def test_init_without_jq_schema_and_missing_dependency(jq_import):
+    converter = JSONConverter(content_key="foo")
+    jq_import.check.assert_not_called()
+    assert converter._jq_schema is None
+    assert converter._content_key == "foo"
+    assert converter._meta_fields is None
+
+
+@patch("haystack.components.converters.json.jq_import")
+def test_init_with_jq_schema_and_missing_dependency(jq_import):
+    jq_import.check.side_effect = ImportError
+    with pytest.raises(ImportError):
+        JSONConverter(jq_schema=".laureates[].motivation")
+
+
+def test_init_with_jq_schema():
+    converter = JSONConverter(jq_schema=".")
+    assert converter._jq_schema == "."
+    assert converter._content_key is None
+    assert converter._meta_fields is None
+
+
+def test_to_dict():
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+
+    assert converter.to_dict() == {
+        "type": "haystack.components.converters.json.JSONConverter",
+        "init_parameters": {
+            "content_key": "motivation",
+            "jq_schema": ".laureates[]",
+            "extra_meta_fields": {"firstname", "surname"},
+        },
+    }
+
+
+def test_from_dict():
+    data = {
+        "type": "haystack.components.converters.json.JSONConverter",
+        "init_parameters": {
+            "content_key": "motivation",
+            "jq_schema": ".laureates[]",
+            "extra_meta_fields": ["firstname", "surname"],
+        },
+    }
+    converter = JSONConverter.from_dict(data)
+
+    assert converter._jq_schema == ".laureates[]"
+    assert converter._content_key == "motivation"
+    assert converter._meta_fields == ["firstname", "surname"]
+
+
+def test_run(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file)}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {}
+
+
+def test_run_with_non_json_file(tmpdir, caplog):
+    test_file = Path(tmpdir / "test_file.md")
+    test_file.write_text("This is not a JSON file.", "utf-8")
+
+    sources = [test_file]
+    converter = JSONConverter(".laureates | .motivation")
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING):
+        result = converter.run(sources=sources)
+
+    records = caplog.records
+    assert len(records) == 1
+    assert (
+        records[0].msg
+        == f"Failed to extract text from {test_file}. Skipping it. Error: parse error: Invalid numeric literal at line 1, column 5"
+    )
+    assert result == {"documents": []}
+
+
+def test_run_with_bad_filter(tmpdir, caplog):
+    test_file = Path(tmpdir / "test_file.json")
+    test_file.write_text(json.dumps(test_data[0]), "utf-8")
+
+    sources = [test_file]
+    converter = JSONConverter(".laureates | .motivation")
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING):
+        result = converter.run(sources=sources)
+
+    records = caplog.records
+    assert len(records) == 1
+    assert (
+        records[0].msg
+        == f'Failed to extract text from {test_file}. Skipping it. Error: Cannot index array with string "motivation"'
+    )
+    assert result == {"documents": []}
+
+
+def test_run_with_single_meta(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    meta = {"creation_date": "1945-05-25T00:00:00"}
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources, meta=meta)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"creation_date": "1945-05-25T00:00:00"}
+
+
+def test_run_with_meta_list(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    meta = [
+        {"creation_date": "1945-05-25T00:00:00"},
+        {"creation_date": "1943-09-03T00:00:00"},
+        {"creation_date": "1989-11-09T00:00:00"},
+    ]
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources, meta=meta)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "creation_date": "1943-09-03T00:00:00"}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file), "creation_date": "1943-09-03T00:00:00"}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"creation_date": "1989-11-09T00:00:00"}
+
+
+def test_run_with_meta_list_of_differing_length(tmpdir):
+    sources = ["random_file.json"]
+
+    meta = [{}, {}]
+    converter = JSONConverter(jq_schema=".")
+    with pytest.raises(ValueError, match="The length of the metadata list must match the number of sources."):
+        converter.run(sources=sources, meta=meta)
+
+
+def test_run_with_jq_schema_and_content_key(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(jq_schema=".laureates[]", content_key="motivation")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content == "who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file)}
+    assert (
+        result["documents"][3].content == "for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {}
+
+
+def test_run_with_jq_schema_content_key_and_extra_meta_fields(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content == "who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "firstname": "Dario", "surname": "Fo"}
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {
+        "file_path": str(second_test_file),
+        "firstname": "Stanley",
+        "surname": "Cohen",
+    }
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {
+        "file_path": str(second_test_file),
+        "firstname": "Rita",
+        "surname": "Levi-Montalcini",
+    }
+    assert (
+        result["documents"][3].content == "for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"firstname": "Enrico", "surname": "Fermi"}
+
+
+def test_run_with_content_key(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(content_key="category")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 3
+    assert result["documents"][0].content == "literature"
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "medicine"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "physics"
+    assert result["documents"][2].meta == {}
+
+
+def test_run_with_content_key_and_extra_meta_fields(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(content_key="category", extra_meta_fields={"year"})
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 3
+    assert result["documents"][0].content == "literature"
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "year": "1997"}
+    assert result["documents"][1].content == "medicine"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "year": "1986"}
+    assert result["documents"][2].content == "physics"
+    assert result["documents"][2].meta == {"year": "1938"}
+
+
+def test_run_with_jq_schema_content_key_and_extra_meta_fields_literal(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(jq_schema=".laureates[]", content_key="motivation", extra_meta_fields="*")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "who emulates the jesters of the Middle Ages in scourging authority and upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {
+        "file_path": str(first_test_file),
+        "id": "674",
+        "firstname": "Dario",
+        "surname": "Fo",
+        "share": "1",
+    }
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {
+        "file_path": str(second_test_file),
+        "id": "434",
+        "firstname": "Stanley",
+        "surname": "Cohen",
+        "share": "2",
+    }
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {
+        "file_path": str(second_test_file),
+        "id": "435",
+        "firstname": "Rita",
+        "surname": "Levi-Montalcini",
+        "share": "2",
+    }
+    assert (
+        result["documents"][3].content
+        == "for his demonstrations of the existence of new radioactive elements produced by neutron irradiation, "
+        "and for his related discovery of nuclear reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"id": "46", "firstname": "Enrico", "surname": "Fermi", "share": "1"}