refactor(docling): add meta parameter to run(); introduce sources; deprecate paths (#3103)

* fix(docling): add meta param to run() and rename paths to sources

* feat(docling): accept ByteStream in sources and use normalize_metadata

* style(docling): apply ruff formatting

* fix(docling): fix mypy type error for deprecated paths parameter

* fix(docling): address reviewer feedback on paths type, meta docstring and source.meta safety

* fix(docling): reorder run() params, fix docstring order, move pytest import to top

* style(docling): fix ruff import sort in test_converter.py

Author: SyedShahmeerAli12

integrations/docling/src/haystack_integrations/components/converters/docling/converter.py

@@ -1,13 +1,17 @@
 """Docling Haystack converter module."""
 
 import json
+import os
+import tempfile
+import warnings
 from abc import ABC, abstractmethod
-from collections.abc import Iterable
 from enum import Enum
 from pathlib import Path
 from typing import Any
 
 from haystack import Document, component
+from haystack.components.converters.utils import normalize_metadata
+from haystack.dataclasses import ByteStream
 
 from docling.chunking import BaseChunk, BaseChunker, HybridChunker
 from docling.datamodel.document import DoclingDocument
@@ -98,46 +102,83 @@ def __init__(
     @component.output_types(documents=list[Document])
     def run(
         self,
-        paths: Iterable[Path | str],
+        paths: list[str | Path] | None = None,
+        sources: list[str | Path | ByteStream] | None = None,
+        meta: dict[str, Any] | list[dict[str, Any]] | None = None,
     ) -> dict[str, list[Document]]:
         """
         Run the DoclingConverter.
 
-        :param paths: The input document locations, either as local paths or URLs.
+        :param paths: Deprecated. Use `sources` instead.
+        :param sources: List of file paths, URLs, or ByteStream objects to convert.
+        :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 a source is a ByteStream, its own metadata is also merged into the output.
         :returns:
             A dictionary with key `"documents"` containing the output Haystack Documents.
+        :raises ValueError: If `meta` is a list whose length does not match the number of sources.
         :raises RuntimeError: If an unexpected `export_type` is encountered.
         """
+        if paths is not None:
+            warnings.warn(
+                "The 'paths' parameter is deprecated. Use 'sources' instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            if sources is None:
+                sources = list(paths)  # type: ignore[arg-type]
+
+        if sources is None:
+            msg = "Either 'sources' or the deprecated 'paths' parameter must be provided."
+            raise ValueError(msg)
+
+        meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
+
         documents: list[Document] = []
-        for filepath in paths:
-            dl_doc = self._converter_instance.convert(
-                source=filepath,
-                **self.convert_kwargs,
-            ).document
+        for source, source_meta in zip(sources, meta_list, strict=True):
+            if isinstance(source, ByteStream):
+                # docling requires a file path; write ByteStream data to a temp file
+                with tempfile.NamedTemporaryFile(delete=False) as tmp:
+                    tmp.write(source.data)
+                    tmp_path = Path(tmp.name)
+                try:
+                    dl_doc = self._converter_instance.convert(source=tmp_path, **self.convert_kwargs).document
+                finally:
+                    os.unlink(tmp_path)
+                # merge ByteStream meta (e.g. file_path, mime_type) with user-supplied meta
+                merged_meta = {**(source.meta or {}), **source_meta}
+            else:
+                dl_doc = self._converter_instance.convert(source=source, **self.convert_kwargs).document
+                merged_meta = source_meta
 
             if self.export_type == ExportType.DOC_CHUNKS:
                 chunk_iter = self._chunker_instance.chunk(dl_doc=dl_doc)
                 hs_docs = [
                     Document(
                         content=self._chunker_instance.contextualize(chunk=chunk),
-                        meta=self._meta_extractor_instance.extract_chunk_meta(chunk=chunk),
+                        meta={**self._meta_extractor_instance.extract_chunk_meta(chunk=chunk), **merged_meta},
                     )
                     for chunk in chunk_iter
                 ]
                 documents.extend(hs_docs)
             elif self.export_type == ExportType.MARKDOWN:
                 hs_doc = Document(
                     content=dl_doc.export_to_markdown(**self.md_export_kwargs),
-                    meta=self._meta_extractor_instance.extract_dl_doc_meta(dl_doc=dl_doc),
+                    meta={**self._meta_extractor_instance.extract_dl_doc_meta(dl_doc=dl_doc), **merged_meta},
                 )
                 documents.append(hs_doc)
             elif self.export_type == ExportType.JSON:
                 hs_doc = Document(
                     content=json.dumps(dl_doc.export_to_dict()),
-                    meta=self._meta_extractor_instance.extract_dl_doc_meta(dl_doc=dl_doc),
+                    meta={**self._meta_extractor_instance.extract_dl_doc_meta(dl_doc=dl_doc), **merged_meta},
                 )
                 documents.append(hs_doc)
             else:
                 err_msg = f"Unexpected export type: {self.export_type}"
                 raise RuntimeError(err_msg)
+
         return {"documents": documents}

integrations/docling/tests/test_converter.py

@@ -1,9 +1,12 @@
 import json
+import warnings
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
+import pytest
 from haystack.core.serialization import component_from_dict, component_to_dict
+from haystack.dataclasses import ByteStream
 
 from haystack_integrations.components.converters.docling import DoclingConverter, ExportType
 
@@ -213,3 +216,153 @@ def test_component_from_dict_custom_params() -> None:
     assert restored.convert_kwargs == {"raises_on_error": False}
     assert restored.export_type == ExportType.JSON
     assert restored.md_export_kwargs == {"image_placeholder": "[img]"}
+
+
+def test_run_with_sources_parameter() -> None:
+    converter_mock = MagicMock()
+    chunker_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    converter_mock.convert.return_value = SimpleNamespace(document="dl-doc")
+    chunker_mock.chunk.return_value = [SimpleNamespace(text="chunk-1")]
+    chunker_mock.contextualize.return_value = "contextualized-chunk-1"
+    meta_extractor_mock.extract_chunk_meta.return_value = {}
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.DOC_CHUNKS,
+        chunker=chunker_mock,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    result = converter.run(sources=["file.pdf"])
+    assert len(result["documents"]) == 1
+
+
+def test_run_paths_deprecated() -> None:
+    converter_mock = MagicMock()
+    chunker_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    converter_mock.convert.return_value = SimpleNamespace(document="dl-doc")
+    chunker_mock.chunk.return_value = [SimpleNamespace(text="chunk-1")]
+    chunker_mock.contextualize.return_value = "contextualized-chunk-1"
+    meta_extractor_mock.extract_chunk_meta.return_value = {}
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.DOC_CHUNKS,
+        chunker=chunker_mock,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        result = converter.run(paths=["file.pdf"])
+
+    assert len(result["documents"]) == 1
+    assert any(issubclass(w.category, DeprecationWarning) and "paths" in str(w.message) for w in caught)
+
+
+def test_run_meta_single_dict_doc_chunks() -> None:
+    converter_mock = MagicMock()
+    chunker_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    converter_mock.convert.side_effect = [
+        SimpleNamespace(document="dl-doc-a"),
+        SimpleNamespace(document="dl-doc-b"),
+    ]
+    chunker_mock.chunk.side_effect = lambda dl_doc: [SimpleNamespace(text=f"chunk-of-{dl_doc}")]
+    chunker_mock.contextualize.side_effect = lambda chunk: chunk.text
+    meta_extractor_mock.extract_chunk_meta.return_value = {"extractor_key": "extractor_val"}
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.DOC_CHUNKS,
+        chunker=chunker_mock,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    result = converter.run(sources=["a.pdf", "b.pdf"], meta={"custom": "value"})
+    documents = result["documents"]
+
+    assert len(documents) == 2
+    for doc in documents:
+        assert doc.meta["custom"] == "value"
+        assert doc.meta["extractor_key"] == "extractor_val"
+
+
+def test_run_meta_list_of_dicts_markdown() -> None:
+    converter_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    dl_doc_a = MagicMock()
+    dl_doc_a.export_to_markdown.return_value = "markdown-a"
+    dl_doc_b = MagicMock()
+    dl_doc_b.export_to_markdown.return_value = "markdown-b"
+
+    converter_mock.convert.side_effect = [
+        SimpleNamespace(document=dl_doc_a),
+        SimpleNamespace(document=dl_doc_b),
+    ]
+    meta_extractor_mock.extract_dl_doc_meta.return_value = {}
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.MARKDOWN,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    result = converter.run(
+        sources=["a.pdf", "b.pdf"],
+        meta=[{"source_id": "doc-a"}, {"source_id": "doc-b"}],
+    )
+    documents = result["documents"]
+
+    assert len(documents) == 2
+    assert documents[0].meta["source_id"] == "doc-a"
+    assert documents[1].meta["source_id"] == "doc-b"
+
+
+def test_run_meta_list_length_mismatch_raises() -> None:
+    converter_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.MARKDOWN,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    with pytest.raises(ValueError):
+        converter.run(sources=["a.pdf", "b.pdf"], meta=[{"x": 1}])
+
+
+def test_run_with_bytestream_source() -> None:
+    converter_mock = MagicMock()
+    meta_extractor_mock = MagicMock()
+
+    dl_doc = MagicMock()
+    dl_doc.export_to_markdown.return_value = "markdown-content"
+    converter_mock.convert.return_value = SimpleNamespace(document=dl_doc)
+    meta_extractor_mock.extract_dl_doc_meta.return_value = {}
+
+    converter = DoclingConverter(
+        converter=converter_mock,
+        export_type=ExportType.MARKDOWN,
+        meta_extractor=meta_extractor_mock,
+    )
+
+    bytestream = ByteStream(data=b"%PDF-1.4 fake pdf content", meta={"file_path": "uploaded.pdf"})
+
+    with patch("os.unlink"):
+        result = converter.run(sources=[bytestream])
+
+    documents = result["documents"]
+    assert len(documents) == 1
+    # ByteStream meta is merged into the output document
+    assert documents[0].meta["file_path"] == "uploaded.pdf"
+    # docling was called with a temp file path, not the ByteStream directly
+    call_args = converter_mock.convert.call_args
+    assert call_args.kwargs["source"] != bytestream