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.
This commit is contained in:
@@ -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",
|
||||
|
||||
65
src/application/ingestion/pipeline.py
Normal file
65
src/application/ingestion/pipeline.py
Normal file
@@ -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,
|
||||
)
|
||||
62
tests/unit/application/test_pipeline.py
Normal file
62
tests/unit/application/test_pipeline.py
Normal file
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user