From 94684d97aeaef1867ff086237022b2713a871ff0 Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Wed, 19 Aug 2026 14:59:51 +0330 Subject: [PATCH] refactor(ingestion): give the pipeline package a single async entry point Why: - The package exposed 8 modules directly, pushing source-type dispatch and the ADR-0017 thread-offload obligation onto every caller. Changes: - Add parse_and_chunk_document as the sole public entry point. - Demote the individual parsers to internal/test-only. --- src/application/ingestion/__init__.py | 12 +++-- src/application/ingestion/pipeline.py | 65 +++++++++++++++++++++++++ tests/unit/application/test_pipeline.py | 62 +++++++++++++++++++++++ 3 files changed, 135 insertions(+), 4 deletions(-) create mode 100644 src/application/ingestion/pipeline.py create mode 100644 tests/unit/application/test_pipeline.py diff --git a/src/application/ingestion/__init__.py b/src/application/ingestion/__init__.py index ffe0058..6a6e400 100644 --- a/src/application/ingestion/__init__.py +++ b/src/application/ingestion/__init__.py @@ -1,9 +1,11 @@ """Document parsing and fixed-size chunking (ADR-0004, ADR-0018). -Everything here is pure and synchronous: no I/O, no ports, no SDK clients -(ADR-0015 reserves ports for external side effects). Parsing and chunking are -blocking CPU work, so callers run them through `anyio.to_thread.run_sync` with -the ingestion `CapacityLimiter` rather than on the event loop (ADR-0017). +`parse_and_chunk_document` is the entry point callers outside this package +should use: it dispatches on source type and owns the +`anyio.to_thread.run_sync` + `CapacityLimiter` offload required by ADR-0017. +The individual parsers and `chunk_document` are pure, synchronous, and +exported mainly for their own unit tests — calling them directly from an +`async def` route or service is the defect ADR-0017 warns about. """ from src.application.ingestion.chunking import chunk_document, chunk_id_for, split_by_tokens @@ -22,6 +24,7 @@ from src.application.ingestion.models import ( StructuralUnit, ) from src.application.ingestion.normalization import normalize_persian_text +from src.application.ingestion.pipeline import parse_and_chunk_document from src.application.ingestion.spreadsheet_parser import parse_csv, parse_xlsx from src.application.ingestion.tokenizer import count_tokens, get_encoder @@ -40,6 +43,7 @@ __all__ = [ "count_tokens", "get_encoder", "normalize_persian_text", + "parse_and_chunk_document", "parse_csv", "parse_docx", "parse_xlsx", diff --git a/src/application/ingestion/pipeline.py b/src/application/ingestion/pipeline.py new file mode 100644 index 0000000..0fcdee2 --- /dev/null +++ b/src/application/ingestion/pipeline.py @@ -0,0 +1,65 @@ +"""The one caller-facing entry point for parsing and chunking (ADR-0017). + +`parse_docx`/`parse_csv`/`parse_xlsx`/`chunk_document` are blocking, pure +functions; calling any of them directly from an `async def` route or service +is the defect ADR-0017 names explicitly ("one large `python-docx` parse would +stall every concurrent request"). `parse_and_chunk_document` is the only +version of this pipeline callers should reach for: it owns source-type +dispatch and the `anyio.to_thread.run_sync` + `CapacityLimiter` offload, so +that obligation cannot be forgotten at a call site. +""" + +import uuid +from functools import partial + +from anyio import CapacityLimiter, to_thread + +from src.application.ingestion.chunking import chunk_document +from src.application.ingestion.docx_parser import parse_docx +from src.application.ingestion.errors import UnsupportedSourceTypeError +from src.application.ingestion.models import Chunk, ParsedDocument +from src.application.ingestion.spreadsheet_parser import parse_csv, parse_xlsx +from src.config import ChunkingSettings + +_PARSERS = {"csv", "xlsx", "docx"} + + +def _parse(data: bytes, source_type: str, settings: ChunkingSettings) -> ParsedDocument: + if source_type == "docx": + return parse_docx(data, settings) + if source_type == "xlsx": + return parse_xlsx(data) + if source_type == "csv": + return parse_csv(data) + raise UnsupportedSourceTypeError(f"'{source_type}' is not an ingestible source type") + + +def _parse_and_chunk( + data: bytes, source_type: str, file_id: uuid.UUID, settings: ChunkingSettings +) -> list[Chunk]: + parsed = _parse(data, source_type, settings) + return chunk_document(parsed, file_id=file_id, settings=settings) + + +async def parse_and_chunk_document( + data: bytes, + *, + source_type: str, + file_id: uuid.UUID, + settings: ChunkingSettings, + limiter: CapacityLimiter, +) -> list[Chunk]: + """Parse and chunk a document off the event loop, bounded by `limiter`. + + Raises `UnsupportedSourceTypeError` (415), `DocumentParseError` (400), or + `ChunkTooLargeError` — see `src/application/ingestion/errors.py`. Callers + map these to status codes; this module carries no HTTP knowledge + (ADR-0015). The `max_chunks_per_file` ceiling (413) is enforced by the + caller, not here — see Phase 4 of plan 001. + """ + if source_type not in _PARSERS: + raise UnsupportedSourceTypeError(f"'{source_type}' is not an ingestible source type") + return await to_thread.run_sync( + partial(_parse_and_chunk, data, source_type, file_id, settings), + limiter=limiter, + ) diff --git a/tests/unit/application/test_pipeline.py b/tests/unit/application/test_pipeline.py new file mode 100644 index 0000000..62fb364 --- /dev/null +++ b/tests/unit/application/test_pipeline.py @@ -0,0 +1,62 @@ +"""Source-type dispatch and thread-offload for `parse_and_chunk_document`.""" + +import uuid + +import pytest +from anyio import CapacityLimiter + +from src.application.ingestion import UnsupportedSourceTypeError, parse_and_chunk_document +from src.config import ChunkingSettings +from tests.support.documents import PROSE_DOCX, load_document + +pytestmark = [pytest.mark.unit, pytest.mark.asyncio] + + +@pytest.fixture +def settings() -> ChunkingSettings: + return ChunkingSettings() + + +@pytest.fixture +def limiter() -> CapacityLimiter: + return CapacityLimiter(2) + + +async def test_parse_and_chunk_document_docx_returns_chunks( + settings: ChunkingSettings, limiter: CapacityLimiter +) -> None: + chunks = await parse_and_chunk_document( + load_document(PROSE_DOCX), + source_type="docx", + file_id=uuid.uuid4(), + settings=settings, + limiter=limiter, + ) + + assert chunks + assert all(chunk.chunk_index == index for index, chunk in enumerate(chunks)) + + +async def test_parse_and_chunk_document_csv_returns_chunks( + settings: ChunkingSettings, limiter: CapacityLimiter +) -> None: + data = b"name,value\nfirst,1\nsecond,2\n" + + chunks = await parse_and_chunk_document( + data, source_type="csv", file_id=uuid.uuid4(), settings=settings, limiter=limiter + ) + + assert chunks + + +async def test_parse_and_chunk_document_unsupported_type_raises( + settings: ChunkingSettings, limiter: CapacityLimiter +) -> None: + with pytest.raises(UnsupportedSourceTypeError): + await parse_and_chunk_document( + b"whatever", + source_type="doc", + file_id=uuid.uuid4(), + settings=settings, + limiter=limiter, + )