Coverage for haystack/components/converters/json.py: 88%
91 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import json
6import os
7from pathlib import Path
8from typing import Any, Literal
10from haystack import component, default_from_dict, default_to_dict, logging
11from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
12from haystack.dataclasses import ByteStream, Document
13from haystack.lazy_imports import LazyImport
15logger = logging.getLogger(__name__)
17with LazyImport("Run 'pip install jq'") as jq_import:
18 import jq
21@component
22class JSONConverter:
23 """
24 Converts one or more JSON files into a text document.
26 ### Usage examples
28 ```python
29 import json
31 from haystack.components.converters import JSONConverter
32 from haystack.dataclasses import ByteStream
34 source = ByteStream.from_string(json.dumps({"text": "This is the content of my document"}))
36 converter = JSONConverter(content_key="text")
37 results = converter.run(sources=[source])
38 documents = results["documents"]
39 print(documents[0].content)
40 # 'This is the content of my document'
41 ```
43 Optionally, you can also provide a `jq_schema` string to filter the JSON source files and `extra_meta_fields`
44 to extract from the filtered data:
46 ```python
47 import json
49 from haystack.components.converters import JSONConverter
50 from haystack.dataclasses import ByteStream
52 data = {
53 "laureates": [
54 {
55 "firstname": "Enrico",
56 "surname": "Fermi",
57 "motivation": "for his demonstrations of the existence of new radioactive elements produced "
58 "by neutron irradiation, and for his related discovery of nuclear reactions brought about by"
59 " slow neutrons",
60 },
61 {
62 "firstname": "Rita",
63 "surname": "Levi-Montalcini",
64 "motivation": "for their discoveries of growth factors",
65 },
66 ],
67 }
68 source = ByteStream.from_string(json.dumps(data))
69 converter = JSONConverter(
70 jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
71 )
73 results = converter.run(sources=[source])
74 documents = results["documents"]
75 print(documents[0].content)
76 # 'for his demonstrations of the existence of new radioactive elements produced by
77 # neutron irradiation, and for his related discovery of nuclear reactions brought
78 # about by slow neutrons'
80 print(documents[0].meta)
81 # {'firstname': 'Enrico', 'surname': 'Fermi'}
83 print(documents[1].content)
84 # 'for their discoveries of growth factors'
86 print(documents[1].meta)
87 # {'firstname': 'Rita', 'surname': 'Levi-Montalcini'}
88 ```
90 """
92 def __init__(
93 self,
94 jq_schema: str | None = None,
95 content_key: str | None = None,
96 extra_meta_fields: set[str] | Literal["*"] | None = None,
97 store_full_path: bool = False,
98 ) -> None:
99 """
100 Creates a JSONConverter component.
102 An optional `jq_schema` can be provided to extract nested data in the JSON source files.
103 See the [official jq documentation](https://jqlang.github.io/jq/) for more info on the filters syntax.
104 If `jq_schema` is not set, whole JSON source files will be used to extract content.
106 Optionally, you can provide a `content_key` to specify which key in the extracted object must
107 be set as the document's content.
109 If both `jq_schema` and `content_key` are set, the component will search for the `content_key` in
110 the JSON object extracted by `jq_schema`. If the extracted data is not a JSON object, it will be skipped.
112 If only `jq_schema` is set, the extracted data must be a scalar value. If it's a JSON object or array,
113 it will be skipped.
115 If only `content_key` is set, the source JSON file must be a JSON object, else it will be skipped.
117 `extra_meta_fields` can either be set to a set of strings or a literal `"*"` string.
118 If it's a set of strings, it must specify fields in the extracted objects that must be set in
119 the extracted documents. If a field is not found, the meta value will be `None`.
120 If set to `"*"`, all fields that are not `content_key` found in the filtered JSON object will
121 be saved as metadata.
123 Initialization will fail if neither `jq_schema` nor `content_key` are set.
125 :param jq_schema:
126 Optional jq filter string to extract content.
127 If not specified, whole JSON object will be used to extract information.
128 :param content_key:
129 Optional key to extract document content.
130 If `jq_schema` is specified, the `content_key` will be extracted from that object.
131 :param extra_meta_fields:
132 An optional set of meta keys to extract from the content.
133 If `jq_schema` is specified, all keys will be extracted from that object.
134 :param store_full_path:
135 If True, the full path of the file is stored in the metadata of the document.
136 If False, only the file name is stored.
137 """
138 self._compiled_filter = None
139 if jq_schema:
140 jq_import.check()
141 self._compiled_filter = jq.compile(jq_schema)
143 self._jq_schema = jq_schema
144 self._content_key = content_key
145 self._meta_fields = extra_meta_fields
146 self._store_full_path = store_full_path
148 if self._compiled_filter is None and self._content_key is None:
149 msg = "No `jq_schema` nor `content_key` specified. Set either or both to extract data."
150 raise ValueError(msg)
152 def to_dict(self) -> dict[str, Any]:
153 """
154 Serializes the component to a dictionary.
156 :returns:
157 Dictionary with serialized data.
158 """
159 return default_to_dict(
160 self,
161 jq_schema=self._jq_schema,
162 content_key=self._content_key,
163 extra_meta_fields=self._meta_fields,
164 store_full_path=self._store_full_path,
165 )
167 @classmethod
168 def from_dict(cls, data: dict[str, Any]) -> "JSONConverter":
169 """
170 Deserializes the component from a dictionary.
172 :param data:
173 Dictionary to deserialize from.
174 :returns:
175 Deserialized component.
176 """
177 return default_from_dict(cls, data)
179 def _get_content_and_meta(self, source: ByteStream) -> list[tuple[str, dict[str, Any]]]:
180 """
181 Utility function to extract text and metadata from a JSON file.
183 :param source:
184 UTF-8 byte stream.
185 :returns:
186 Collection of text and metadata dict tuples, each corresponding
187 to a different document.
188 """
189 try:
190 file_content = source.data.decode("utf-8")
191 except UnicodeError as exc:
192 logger.warning(
193 "Failed to extract text from {source}. Skipping it. Error: {error}",
194 source=source.meta.get("file_path", "unknown"),
195 error=exc,
196 )
197 return []
199 meta_fields = self._meta_fields or set()
201 if self._compiled_filter is not None:
202 try:
203 objects = list(self._compiled_filter.input_text(file_content))
204 except Exception as exc:
205 logger.warning(
206 "Failed to extract text from {source}. Skipping it. Error: {error}",
207 source=source.meta.get("file_path", "unknown"),
208 error=exc,
209 )
210 return []
211 else:
212 # We just load the whole file as JSON if the user didn't provide a jq filter.
213 # We put it in a list even if it's not to ease handling it later on.
214 try:
215 objects = [json.loads(file_content)]
216 except json.JSONDecodeError as exc:
217 logger.warning(
218 "Failed to extract text from {source}. Skipping it. Error: {error}",
219 source=source.meta.get("file_path", "unknown"),
220 error=exc,
221 )
222 return []
224 result = []
225 if self._content_key is not None:
226 for obj in objects:
227 if not isinstance(obj, dict):
228 logger.warning("Expected a dictionary but got {obj}. Skipping it.", obj=obj)
229 continue
230 if self._content_key not in obj:
231 logger.warning(
232 "'{content_key}' not found in {obj}. Skipping it.", content_key=self._content_key, obj=obj
233 )
234 continue
236 text = obj[self._content_key]
237 if isinstance(text, (dict, list)):
238 logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
239 continue
241 meta = {}
242 if meta_fields == "*":
243 meta = {k: v for k, v in obj.items() if k != self._content_key}
244 else:
245 for field in meta_fields:
246 meta[field] = obj.get(field, None)
247 result.append((text, meta))
248 else:
249 for obj in objects:
250 if isinstance(obj, (dict, list)):
251 logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
252 continue
253 result.append((str(obj), {}))
255 return result
257 @component.output_types(documents=list[Document])
258 def run(
259 self, sources: list[str | Path | ByteStream], meta: dict[str, Any] | list[dict[str, Any]] | None = None
260 ) -> dict[str, Any]:
261 """
262 Converts a list of JSON files to documents.
264 :param sources:
265 A list of file paths or ByteStream objects.
266 :param meta:
267 Optional metadata to attach to the documents.
268 This value can be either a list of dictionaries or a single dictionary.
269 If it's a single dictionary, its content is added to the metadata of all produced documents.
270 If it's a list, the length of the list must match the number of sources.
271 If `sources` contain ByteStream objects, their `meta` will be added to the output documents.
273 :returns:
274 A dictionary with the following keys:
275 - `documents`: A list of created documents.
276 """
277 documents = []
278 meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
280 for source, metadata in zip(sources, meta_list, strict=True):
281 try:
282 bytestream = get_bytestream_from_source(source)
283 except Exception as exc:
284 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=exc)
285 continue
287 data = self._get_content_and_meta(bytestream)
289 for text, extra_meta in data:
290 merged_metadata = {**bytestream.meta, **metadata, **extra_meta}
292 if not self._store_full_path and (file_path := bytestream.meta.get("file_path")):
293 merged_metadata["file_path"] = os.path.basename(file_path)
294 document = Document(content=text, meta=merged_metadata)
295 documents.append(document)
297 return {"documents": documents}