Coverage for haystack/components/converters/csv.py: 95%
87 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import csv
6import io
7import os
8from pathlib import Path
9from typing import Any, Literal
11from haystack import Document, component, logging
12from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
13from haystack.dataclasses import ByteStream
15logger = logging.getLogger(__name__)
17_ROW_MODE_SIZE_WARN_BYTES = 5 * 1024 * 1024 # ~5MB; warn when parsing rows might be memory-heavy
20@component
21class CSVToDocument:
22 """
23 Converts CSV files to Documents.
25 By default, it uses UTF-8 encoding (`utf-8-sig`, which also strips a byte order mark if
26 present) when converting files but
27 you can also set a custom encoding.
28 It can attach metadata to the resulting documents.
30 ### Usage example
31 ```python
32 from haystack.components.converters.csv import CSVToDocument
33 from datetime import datetime
35 converter = CSVToDocument()
36 results = converter.run(
37 sources=["test/test_files/csv/sample_1.csv"], meta={"date_added": datetime.now().isoformat()}
38 )
39 documents = results["documents"]
41 print(documents[0].content)
42 # >> 'col1,col2\\nrow1,row1\\nrow2,row2\\n'
43 ```
44 """
46 def __init__(
47 self,
48 encoding: str = "utf-8-sig",
49 store_full_path: bool = False,
50 *,
51 conversion_mode: Literal["file", "row"] = "file",
52 delimiter: str = ",",
53 quotechar: str = '"',
54 ) -> None:
55 """
56 Creates a CSVToDocument component.
58 :param encoding:
59 The encoding of the csv files to convert.
60 If the encoding is specified in the metadata of a source ByteStream,
61 it overrides this value.
62 :param store_full_path:
63 If True, the full path of the file is stored in the metadata of the document.
64 If False, only the file name is stored.
65 :param conversion_mode:
66 - "file" (default): one Document per CSV file whose content is the raw CSV text.
67 - "row": convert each CSV row to its own Document (requires `content_column` in `run()`).
68 :param delimiter:
69 CSV delimiter used when parsing in row mode (passed to ``csv.DictReader``).
70 :param quotechar:
71 CSV quote character used when parsing in row mode (passed to ``csv.DictReader``).
72 """
73 self.encoding = encoding
74 self.store_full_path = store_full_path
75 self.conversion_mode = conversion_mode
76 self.delimiter = delimiter
77 self.quotechar = quotechar
79 # Basic validation
80 if len(self.delimiter) != 1:
81 raise ValueError("CSVToDocument: delimiter must be a single character.")
82 if len(self.quotechar) != 1:
83 raise ValueError("CSVToDocument: quotechar must be a single character.")
85 @component.output_types(documents=list[Document])
86 def run(
87 self,
88 sources: list[str | Path | ByteStream],
89 *,
90 content_column: str | None = None,
91 meta: dict[str, Any] | list[dict[str, Any]] | None = None,
92 ) -> dict[str, Any]:
93 """
94 Converts CSV files to a Document (file mode) or to one Document per row (row mode).
96 :param sources:
97 List of file paths or ByteStream objects.
98 :param content_column:
99 **Required when** ``conversion_mode="row"``.
100 The column name whose values become ``Document.content`` for each row.
101 The column must exist in the CSV header.
102 :param meta:
103 Optional metadata to attach to the documents.
104 This value can be either a list of dictionaries or a single dictionary.
105 If it's a single dictionary, its content is added to the metadata of all produced documents.
106 If it's a list, the length of the list must match the number of sources, because the two lists will
107 be zipped.
108 If `sources` contains ByteStream objects, their `meta` will be added to the output documents.
109 :returns:
110 A dictionary with the following keys:
111 - `documents`: Created documents
112 """
113 documents: list[Document] = []
115 meta_list = normalize_metadata(meta, sources_count=len(sources))
117 for source, metadata in zip(sources, meta_list, strict=True):
118 try:
119 bytestream = get_bytestream_from_source(source)
120 except Exception as e:
121 logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=e)
122 continue
124 try:
125 encoding = bytestream.meta.get("encoding", self.encoding)
126 raw = io.BytesIO(bytestream.data).getvalue()
127 data = raw.decode(encoding=encoding)
128 except Exception as e:
129 logger.warning(
130 "Could not convert file {source}. Skipping it. Error message: {error}", source=source, error=e
131 )
132 continue
134 merged_metadata = {**bytestream.meta, **metadata}
136 if not self.store_full_path and "file_path" in bytestream.meta:
137 file_path = bytestream.meta.get("file_path")
138 if file_path: # Ensure the value is not None for mypy
139 merged_metadata["file_path"] = os.path.basename(file_path)
141 # Mode: file (backward-compatible default) -> one Document per file
142 if self.conversion_mode == "file":
143 documents.append(Document(content=data, meta=merged_metadata))
144 continue
146 # --- ROW MODE (strict) ---
147 # Require content_column in run(); no fallback
148 if not content_column:
149 raise ValueError(
150 "CSVToDocument(row): 'content_column' is required in run() when conversion_mode='row'."
151 )
153 # Warn for large CSVs in row mode (memory consideration)
154 try:
155 size_bytes = len(raw)
156 if size_bytes > _ROW_MODE_SIZE_WARN_BYTES:
157 logger.warning(
158 "CSVToDocument(row): parsing a large CSV (~{mb:.1f} MB). "
159 "Consider chunking/streaming if you hit memory issues.",
160 mb=size_bytes / (1024 * 1024),
161 )
162 except Exception:
163 pass
165 # Create DictReader; if this fails, raise (no fallback)
166 try:
167 # ``restkey`` ensures surplus fields on ragged rows (rows with more values than the
168 # header, e.g. an unquoted comma inside a value) land under an explicit string key
169 # instead of the default ``None`` key, which would break ``Document`` id generation.
170 reader = csv.DictReader(
171 io.StringIO(data), delimiter=self.delimiter, quotechar=self.quotechar, restkey="extra_columns"
172 )
173 except Exception as e:
174 raise RuntimeError(f"CSVToDocument(row): could not parse CSV rows for {source}: {e}") from e
176 # Validate header contains content_column; strict error if missing
177 header = reader.fieldnames or []
178 if content_column not in header:
179 raise ValueError(
180 f"CSVToDocument(row): content_column='{content_column}' not found in header "
181 f"for {source}. Available columns: {header}"
182 )
184 # Build documents; if a row processing fails, raise immediately (no skip)
185 for i, row in enumerate(reader):
186 try:
187 doc = self._build_document_from_row(
188 row=row, base_meta=merged_metadata, row_index=i, content_column=content_column
189 )
190 except Exception as e:
191 raise RuntimeError(f"CSVToDocument(row): failed to process row {i} for {source}: {e}") from e
192 documents.append(doc)
194 return {"documents": documents}
196 # ----- helpers -----
197 def _safe_value(self, value: Any) -> str:
198 """Normalize CSV cell values: None -> '', everything -> str."""
199 return "" if value is None else str(value)
201 def _build_document_from_row(
202 self, row: dict[str, Any], base_meta: dict[str, Any], row_index: int, content_column: str
203 ) -> Document:
204 """
205 Build a ``Document`` from one parsed CSV row.
207 :param row: Mapping of column name to cell value for the current row
208 (as produced by ``csv.DictReader``).
209 :param base_meta: File-level and user-provided metadata to start from
210 (for example: ``file_path``, ``encoding``).
211 :param row_index: Zero-based row index in the CSV; stored as
212 ``row_number`` in the output document's metadata.
213 :param content_column: Column name to use for ``Document.content``.
214 :returns: A ``Document`` with chosen content and merged metadata.
215 Remaining row columns are added to ``meta`` with collision-safe
216 keys (prefixed with ``csv_`` if needed).
217 """
218 row_meta = {**base_meta, "row_number": row_index}
220 # content (strict: content_column must exist; validated by caller)
221 content = self._safe_value(row.get(content_column))
223 # merge remaining columns into meta with collision handling
224 for k, v in row.items():
225 if k == content_column:
226 continue
227 key_to_use = k
228 if key_to_use in row_meta:
229 # Avoid clobbering existing meta like file_path/encoding; prefix and de-dupe
230 base_key = f"csv_{key_to_use}"
231 key_to_use = base_key
232 suffix = 1
233 while key_to_use in row_meta:
234 key_to_use = f"{base_key}_{suffix}"
235 suffix += 1
236 row_meta[key_to_use] = self._safe_value(v)
238 return Document(content=content, meta=row_meta)