From 6a0e71a30d8085542e095f10f4093c19d9240b1f Mon Sep 17 00:00:00 2001 From: Richie Cahill Date: Thu, 16 Jul 2026 11:57:49 -0400 Subject: [PATCH] feat(ebook-search): implement async file path resolution and EPUB discovery --- python/ebook_search/ingest.py | 44 +++++++++++++++++---------------- tests/ebook_search/test_core.py | 35 +++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 22 deletions(-) diff --git a/python/ebook_search/ingest.py b/python/ebook_search/ingest.py index 8d0e6c0..0413a91 100644 --- a/python/ebook_search/ingest.py +++ b/python/ebook_search/ingest.py @@ -11,6 +11,7 @@ from pathlib import Path from typing import TYPE_CHECKING import tiktoken +from anyio import Path as AsyncPath from sqlalchemy import or_, select from python.ebook_search.epub_parse import parse_epub @@ -74,18 +75,18 @@ def chunk_text( return [chunk for chunk in chunks if chunk.text] -def find_library_epubs(library_path: str) -> tuple[Path, list[Path] | None]: - """Resolve one configured library path and collect its EPUB files (blocking filesystem walk). +async def find_library_epubs(library_path: str) -> tuple[AsyncPath, list[AsyncPath] | None]: + """Resolve one configured library path and collect its EPUB files asynchronously. Returns: tuple[Path, list[Path] | None]: The expanded path and its EPUB files, or ``None`` when the path is neither an EPUB file nor a directory. """ - path = Path(library_path).expanduser() - if path.is_file() and path.suffix.lower() == ".epub": + path = await AsyncPath(library_path).expanduser() + if await path.is_file() and path.suffix.lower() == ".epub": return path, [path] - if path.is_dir(): - return path, sorted(path.rglob("*.epub")) + if await path.is_dir(): + return path, sorted([epub_path async for epub_path in path.rglob("*.epub")]) return path, None @@ -93,7 +94,7 @@ async def ingest_configured_paths(session: AsyncSession, config: EbookSearchConf """Ingest every EPUB found under configured library paths.""" count = 0 for library_path in config.library_paths: - path, epub_paths = await asyncio.to_thread(find_library_epubs, library_path) + path, epub_paths = await find_library_epubs(library_path) logger.info(f"ebook_ingest_path_start {path=}") if epub_paths is None: logger.warning(f"ebook_ingest_path_missing {path=}") @@ -104,20 +105,21 @@ async def ingest_configured_paths(session: AsyncSession, config: EbookSearchConf return count -def resolve_ingest_path(path: Path) -> Path: - """Expand and resolve an ingest path (blocking filesystem call).""" - return path.expanduser().resolve() +async def resolve_ingest_path(path: Path | AsyncPath) -> AsyncPath: + """Expand and resolve an ingest path without blocking the event loop.""" + expanded_path = await AsyncPath(path).expanduser() + return await expanded_path.resolve() -async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConfig) -> bool: +async def ingest_file(session: AsyncSession, path: Path | AsyncPath, config: EbookSearchConfig) -> bool: """Ingest one EPUB file. Return True when the database changed.""" try: - resolved_path = await asyncio.to_thread(resolve_ingest_path, path) + resolved_path = await resolve_ingest_path(path) logger.info(f"ebook_ingest_file_start {resolved_path=}") - file_hash = await asyncio.to_thread(sha256_file, resolved_path) + file_hash = await sha256_file(resolved_path) existing = await find_existing_source(session, resolved_path, file_hash) if existing is not None and existing.file_sha256 == file_hash: - stat = resolved_path.stat() + stat = await resolved_path.stat() existing.file_path = str(resolved_path) existing.file_mtime = datetime.fromtimestamp(stat.st_mtime, tz=UTC) existing.file_size = stat.st_size @@ -129,8 +131,8 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf await session.delete(existing) await session.flush() - stat = resolved_path.stat() - parsed = await asyncio.to_thread(parse_epub, resolved_path) + stat = await resolved_path.stat() + parsed = await asyncio.to_thread(parse_epub, Path(resolved_path)) source = EbookSource( title=parsed.title, author=parsed.author, @@ -170,7 +172,7 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf return True -async def find_existing_source(session: AsyncSession, path: Path, file_hash: str) -> EbookSource | None: +async def find_existing_source(session: AsyncSession, path: Path | AsyncPath, file_hash: str) -> EbookSource | None: """Find an existing source by canonical path or file hash.""" return await session.scalar( select(EbookSource).where(or_(EbookSource.file_path == str(path), EbookSource.file_sha256 == file_hash)) @@ -209,10 +211,10 @@ def add_chapter_chunks( return chunk_index -def sha256_file(path: Path) -> str: - """Calculate the SHA-256 digest for a file.""" +async def sha256_file(path: AsyncPath) -> str: + """Calculate the SHA-256 digest for a file without blocking the event loop.""" digest = hashlib.sha256() - with path.open("rb") as file: - for block in iter(lambda: file.read(1024 * 1024), b""): + async with await path.open("rb") as file: + while block := await file.read(1024 * 1024): digest.update(block) return digest.hexdigest() diff --git a/tests/ebook_search/test_core.py b/tests/ebook_search/test_core.py index 7d52c61..0b75280 100644 --- a/tests/ebook_search/test_core.py +++ b/tests/ebook_search/test_core.py @@ -27,7 +27,13 @@ from python.ebook_search.bm25_corpus import ( ) from python.ebook_search.config import EbookSearchConfig, RerankConfig, load_config from python.ebook_search.embeddings import MODEL_DIMENSIONS, ensure_embedding_models -from python.ebook_search.ingest import chunk_text, find_existing_source +from python.ebook_search.ingest import ( + chunk_text, + find_existing_source, + find_library_epubs, + resolve_ingest_path, + sha256_file, +) from python.ebook_search.search import ( SearchResponse, SearchResult, @@ -106,6 +112,33 @@ async def test_find_existing_source_matches_path_or_hash() -> None: assert await find_existing_source(session, Path("/new/book.epub"), "a" * 64) == source +async def test_async_ingest_path_resolution_and_hashing(tmp_path: Path) -> None: + """Ingest file metadata uses AnyIO's asynchronous path operations.""" + path = tmp_path / "book.epub" + path.write_bytes(b"epub content") + + resolved_path = await resolve_ingest_path(path) + + assert resolved_path == await resolved_path.resolve() + assert await sha256_file(resolved_path) == "4b71fc6fc452ae1ff7832bfe374ae56ab4d0649acf6af8ab4d068d858ea60448" + + +async def test_find_library_epubs_uses_async_path_walk(tmp_path: Path) -> None: + """Library discovery finds EPUB files through AnyIO's async path API.""" + (tmp_path / "nested").mkdir() + first_epub = tmp_path / "first.epub" + second_epub = tmp_path / "nested" / "second.epub" + first_epub.write_bytes(b"first") + second_epub.write_bytes(b"second") + (tmp_path / "not-a-book.txt").write_text("ignore", encoding="utf-8") + + path, epub_paths = await find_library_epubs(str(tmp_path)) + + assert str(path) == str(tmp_path) + assert epub_paths is not None + assert [str(epub_path) for epub_path in epub_paths] == [str(first_epub), str(second_epub)] + + async def test_bm25_corpus_uses_existing_search_text_without_duplicate_metadata() -> None: engine = await build_async_engine() async with AsyncSession(engine, expire_on_commit=False) as session: