feat(04-02): convert EmlLoader, DocxLoader, and DocLoader to non-blocking I/O

- EmlLoader: Wrap file reads, temp file ops, and cleanup with asyncio.to_thread
  - Created _read_file_bytes and _write_temp_file sync helpers
  - Wrap os.path.exists and os.unlink in finally block
  - Replace manual file write with await self.save_content()
- DocxLoader: Wrap MarkItDown converter and zipfile extraction with asyncio.to_thread
  - await asyncio.to_thread for converter.convert and get_images_from_zip
  - Updated save_content call to use await
- DocLoader: Wrap Spire.Doc operations in thread pool with proper cleanup
  - Created _convert_doc_to_docx sync helper
  - Wrap conversion and os.remove with asyncio.to_thread
  - Proper try/finally for temp file cleanup

Author: paultranvan

openrag/components/indexer/loaders/doc.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import tempfile
 
@@ -14,16 +15,22 @@ def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
         self.MDLoader = DocxLoader(**kwargs)
 
-    async def aload_document(self, file_path, metadata, save_markdown=False):
-        """Here we convert the document to docx format, save it in local and then use the MarkItDownLoader
-        to convert it to markdown."""
+    def _convert_doc_to_docx(self, file_path):
+        """Convert .doc to .docx using Spire.Doc (blocking operations in thread pool)."""
         document = Document()
         document.LoadFromFile(str(file_path))
-        # file_path = "converted/sample2.docx"
         with tempfile.NamedTemporaryFile(delete=False, suffix=".docx") as temp_file:
-            file_path = temp_file.name
-            document.SaveToFile(file_path, FileFormat.Docx2016)
-        result_string = await self.MDLoader.aload_document(file_path, metadata, save_markdown)
-        os.remove(file_path)
-        document.Close()
+            temp_path = temp_file.name
+            document.SaveToFile(temp_path, FileFormat.Docx2016)
+        return document, temp_path
+
+    async def aload_document(self, file_path, metadata, save_markdown=False):
+        """Here we convert the document to docx format, save it in local and then use the MarkItDownLoader
+        to convert it to markdown."""
+        document, temp_path = await asyncio.to_thread(self._convert_doc_to_docx, file_path)
+        try:
+            result_string = await self.MDLoader.aload_document(temp_path, metadata, save_markdown)
+        finally:
+            await asyncio.to_thread(os.remove, temp_path)
+            document.Close()
         return result_string

openrag/components/indexer/loaders/docx.py

@@ -1,3 +1,4 @@
+import asyncio
 import re
 import zipfile
 from io import BytesIO
@@ -28,11 +29,12 @@ def __init__(self, **kwargs):
         self.converter = MarkItDown()
 
     async def aload_document(self, file_path, metadata, save_markdown=False):
-        result = self.converter.convert(file_path).text_content
+        convert_result = await asyncio.to_thread(self.converter.convert, file_path)
+        result = convert_result.text_content
 
         if self.image_captioning:
             # Handle embedded images (extracted from docx zip)
-            images = self.get_images_from_zip(file_path)
+            images = await asyncio.to_thread(self.get_images_from_zip, file_path)
             captions = await self.caption_images(images, desc="Captioning embedded images")
             for caption in captions:
                 result = re.sub(
@@ -54,7 +56,7 @@ async def aload_document(self, file_path, metadata, save_markdown=False):
 
         doc = Document(page_content=result, metadata=metadata)
         if save_markdown:
-            self.save_content(result, str(file_path))
+            await self.save_content(result, str(file_path))
         return doc
 
     def get_images_from_zip(self, input_file):

openrag/components/indexer/loaders/eml_loader.py

@@ -1,3 +1,4 @@
+import asyncio
 import datetime
 import email
 import email.errors
@@ -28,10 +29,20 @@ def __init__(self, **kwargs):
         # Get available loaders for processing attachments
         self.loader_classes = get_loader_classes(config=self.config)
 
+    def _read_file_bytes(self, file_path):
+        """Read file bytes (blocking operation for thread pool)."""
+        with open(file_path, "rb") as fhdl:
+            return fhdl.read()
+
+    def _write_temp_file(self, suffix, data):
+        """Write temp file (blocking operation for thread pool)."""
+        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
+            temp_file.write(data)
+            return temp_file.name
+
     async def aload_document(self, file_path, metadata: dict | None = None, save_markdown: bool = False):
         try:
-            with open(file_path, "rb") as fhdl:
-                raw_email = fhdl.read()
+            raw_email = await asyncio.to_thread(self._read_file_bytes, file_path)
 
             # Parse email using standard email library
             email_msg = email.message_from_bytes(raw_email)
@@ -131,9 +142,9 @@ async def aload_document(self, file_path, metadata: dict | None = None, save_mar
 
                             if loader_cls:
                                 # Save attachment to temporary file
-                                with tempfile.NamedTemporaryFile(suffix=file_ext, delete=False) as temp_file:
-                                    temp_file.write(attachment["raw"])
-                                    temp_file_path = temp_file.name
+                                temp_file_path = await asyncio.to_thread(
+                                    self._write_temp_file, file_ext, attachment["raw"]
+                                )
 
                                 try:
                                     # Use appropriate loader to process attachment
@@ -236,8 +247,8 @@ async def aload_document(self, file_path, metadata: dict | None = None, save_mar
                                             attachments_text += f"Text fallback failed: {str(text_e)[:100]}...\n"
                                 finally:
                                     # Clean up temporary file
-                                    if os.path.exists(temp_file_path):
-                                        os.unlink(temp_file_path)
+                                    if await asyncio.to_thread(os.path.exists, temp_file_path):
+                                        await asyncio.to_thread(os.unlink, temp_file_path)
 
                             # Special handling for images with captioning if no specific loader or captioning is enabled
                             elif file_ext in [
@@ -321,10 +332,8 @@ async def aload_document(self, file_path, metadata: dict | None = None, save_mar
 
             # Save content body to a file if requested
             if save_markdown:
-                markdown_path = Path(file_path).with_suffix(".md")
-                with open(markdown_path, "w", encoding="utf-8") as md_file:
-                    md_file.write(content_body)
-                metadata["markdown_path"] = str(markdown_path)
+                await self.save_content(content_body, str(file_path))
+                metadata["markdown_path"] = str(Path(file_path).with_suffix(".md"))
         except OSError as e:
             # File I/O error reading email file
             raise ValueError(f"Cannot read email file: {e}")