Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/loaders/_io_utils.py: 44%

1"""Shared async file I/O helpers for document loaders.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from pathlib import Path 

7 

8try: 

9 import aiofiles 

10 

11 HAS_AIOFILES = True 

12except ImportError: 

13 HAS_AIOFILES = False 

14 

15 

16async def read_file_bytes(path: Path) -> bytes: 

17 """Read a file as bytes, using aiofiles when available.""" 

18 if HAS_AIOFILES: 

19 async with aiofiles.open(path, "rb") as file_handle: 

20 return await file_handle.read() 

21 return await asyncio.to_thread(path.read_bytes) 

22 

23 

24async def read_file_text(path: Path, encoding: str = "utf-8") -> str: 

25 """Read a file as text, using aiofiles when available.""" 

26 if HAS_AIOFILES: 

27 async with aiofiles.open(path, encoding=encoding) as file_handle: 

28 return await file_handle.read() 

29 return await asyncio.to_thread(path.read_text, encoding=encoding)