Pipeline

domain.fetch_pipeline.utils.pipeline

Summary

Pipeline orchestration for fetch requests.

Drives extraction stages (pdf, email, tabular, llama), persists partial output on every stage, pauses on ambiguity, enriches raw expenses, saves expense documents, and emits SSE progress events. The orchestrator depends on injected services and the fetch pipeline ports.

---

Notes

Guarantees:

• The orchestrator never touches I/O infrastructure directly; fetch, persistence, enrichment and messaging all flow through the injected services and ports.

Lifecycle:

• A pause re-raises PauseForResolution after saving the paused request state; the caller decides whether to resume later.

Classes

PipelineOrchestrator

PipelineOrchestrator(
    fr_svc,
    ambiguity_svc,
    expense_svc,
    pdf_fetcher: PdfFetcher,
    email_fetcher: EmailFetcher,
    xlsx_fetcher: XlsxFetcher,
    csv_fetcher: CsvFetcher,
    sse: ProgressEventBus,
    pause_on_ambiguity: bool = True,
    resolver: Optional[Callable] = None,
    resolved_choices: Optional[dict] = None,
    entity_svc=None,
    tag_svc=None,
    cache_store=None,
)

Orchestrate extraction, enrichment and persistence for a fetch request.

Drives the extraction root over fetched content, persists partial stage output, pauses on ambiguity, enriches raw expenses, saves expense documents through the injected services, and streams progress events via the SSE bus.

Attributes:

Name	Type	Description
fr_svc		Fetch request service used to read and save the request.
ambiguity_svc		Service persisting ambiguity records.
expense_svc		Service creating expense documents.
pdf_fetcher	PdfFetcher	Fetcher for PDF sources.
email_fetcher	EmailFetcher	Fetcher for email sources.
xlsx_fetcher	XlsxFetcher	Fetcher for xlsx sources.
csv_fetcher	CsvFetcher	Fetcher for csv sources.
sse	ProgressEventBus	Bus streaming progress events.
pause_on_ambiguity	bool	Whether ambiguity is paused for resolution.
resolver		Optional ambiguity resolver callback.
resolved_choices	dict	Pre-chosen ambiguity resolutions.
entity_svc		Optional entity service used for expense creation.
tag_svc		Optional tag service used for expense creation.
cache_store		Optional read-cache store invalidated after saving.

Functions

run_pipeline async

run_pipeline(fr_id: str) -> None

Run the full fetch pipeline for a request.

Reads the request, builds the extractor, fetches content, extracts raw expenses, enriches them, and persists expense documents.

Parameters:

Name	Type	Description	Default
fr_id	str	Fetch request id.	required

Returns:

Name	Type	Description
None	None	Nothing.

Notes

An ambiguity pause is re-raised as PauseForResolution after the paused state has been saved.

Functions