Progress API

scinr.newton.progress.ProgressEvent dataclass

One progress event emitted by :func:~scinr.newton.pipeline.run_pipeline.

Only the fields relevant to kind are set; every other field is None. The event never carries the document itself nor any extracted content. Use :meth:to_dict to serialize it (e.g. to JSON).

Attributes:

Name	Type	Description
kind	ProgressEventKind	What happened (see the "Event reference" table in the Progress Tracking guide).
timestamp	datetime	When the event was emitted (timezone-aware, UTC).
job_id	str | None	The job_id passed to run_pipeline().
tenant_id	str | None	The document's tenant for document events, otherwise the tenant_id passed to run_pipeline() (None = public).
document_index	int | None	Position of the document in the list of discovered documents; unique within one run_pipeline() call. None for pipeline events and for stage="tabular".
doc_path	str | None	Path of the document; stable for the whole life of the document.
document_name	str | None	Current document name — the name predicted from the file at first, the real document_name once extraction/ingestion produced it.
source_path	str | None	File the document was discovered from, or None.
stage	str | None	Pipeline stage of a stage_* event: "preprocess", "extraction", "ingestion", "annotation", "entity_extraction" or "tabular".
stages	tuple[str, ...] | None	pipeline_started: the stages requested; documents_discovered: the document stages that will actually run.
planned_stages	tuple[str, ...] | None	document_started: the stages this document will run.
total_documents	int | None	documents_discovered: number of documents to process.
duration_s	float | None	Seconds spent in the stage / document / pipeline that just ended.
metrics	Mapping[str, int] | None	stage_completed / stage_failed: counters of the stage (nodes_processed / nodes_failed for document stages; total_processed / total_failed / documents for stage="tabular"). Read-only.
status	DocumentStatus | None	document_finished: "success", "partial", "failed" or "skipped".
stopped_at	str | None	document_finished: stage where the document stopped because of a failure, or None.
success	bool | None	pipeline_completed: same value as PipelineResult.success.
error	str | None	stage_failed / document_finished / pipeline_failed: error message, with credentials redacted.

to_dict

to_dict() -> dict[str, Any]

Return the event as a plain, JSON-serializable dict.

timestamp becomes an ISO 8601 string, metrics a dict and the stage tuples lists. (dataclasses.asdict() cannot copy the read-only metrics mapping — use this method instead.)

scinr.newton.progress.ProgressCallback

Base class for synchronous progress callbacks.

Subclass it and override either :meth:on_event (one entry point for every event) or only the on_<kind> methods you care about; every method is a no-op by default.

Runs in a dedicated worker thread (one per run_pipeline() call), never on the event loop, so it may block — call a blocking database client, requests, etc. Events are delivered one at a time and in order, so within one run_pipeline() call the callback never runs concurrently with itself. A single instance shared by several concurrent run_pipeline() calls can run concurrently with itself: do not assume thread-safety in that case.

Exceptions raised by a callback are logged (logger scinr.newton.progress) and ignored; they never abort the ingestion.

on_document_finished

on_document_finished(event: ProgressEvent) -> None

A document finished processing (status, stopped_at, error).

on_document_started

on_document_started(event: ProgressEvent) -> None

A document started processing (planned_stages).

on_documents_discovered

on_documents_discovered(event: ProgressEvent) -> None

The documents to process are known (total_documents, stages).

on_event

on_event(event: ProgressEvent) -> None

Called for every event. By default dispatches to on_<event.kind>.

on_pipeline_completed

on_pipeline_completed(event: ProgressEvent) -> None

The run returned a PipelineResult (always the last event).

on_pipeline_failed

on_pipeline_failed(event: ProgressEvent) -> None

The run raised an exception (always the last event).

on_pipeline_started

on_pipeline_started(event: ProgressEvent) -> None

The run started (always the first event).

on_stage_completed

on_stage_completed(event: ProgressEvent) -> None

A document (or the tabular batch) finished stage (metrics).

on_stage_failed

on_stage_failed(event: ProgressEvent) -> None

A document stopped at stage because of a failure (error).

on_stage_started

on_stage_started(event: ProgressEvent) -> None

A document (or the tabular batch) entered stage.

scinr.newton.progress.AsyncProgressCallback

Base class for asynchronous progress callbacks.

Same contract as :class:ProgressCallback, but every method is a coroutine awaited on the pipeline's own event loop — so it can use async clients bound to that loop (motor, httpx, the async Neo4j driver, …). Events are delivered one at a time and in order; within one run_pipeline() call the callback never runs concurrently with itself (a shared instance across concurrent calls can).

A coroutine that blocks internally (requests, pymongo, time.sleep) freezes the whole event loop — and with it the ingestion. Write blocking code as a :class:ProgressCallback instead.

Exceptions raised by a callback are logged (logger scinr.newton.progress) and ignored; they never abort the ingestion.

on_document_finished async

on_document_finished(event: ProgressEvent) -> None

A document finished processing (status, stopped_at, error).

on_document_started async

on_document_started(event: ProgressEvent) -> None

A document started processing (planned_stages).

on_documents_discovered async

on_documents_discovered(event: ProgressEvent) -> None

The documents to process are known (total_documents, stages).

on_event async

on_event(event: ProgressEvent) -> None

Called for every event. By default dispatches to on_<event.kind>.

on_pipeline_completed async

on_pipeline_completed(event: ProgressEvent) -> None

The run returned a PipelineResult (always the last event).

on_pipeline_failed async

on_pipeline_failed(event: ProgressEvent) -> None

The run raised an exception (always the last event).

on_pipeline_started async

on_pipeline_started(event: ProgressEvent) -> None

The run started (always the first event).

on_stage_completed async

on_stage_completed(event: ProgressEvent) -> None

A document (or the tabular batch) finished stage (metrics).

on_stage_failed async

on_stage_failed(event: ProgressEvent) -> None

A document stopped at stage because of a failure (error).

on_stage_started async

on_stage_started(event: ProgressEvent) -> None

A document (or the tabular batch) entered stage.

scinr.newton.progress.ProgressTracker

Built-in progress callback that keeps an aggregated, in-memory view of a run.

Pass it in run_pipeline(callbacks=[tracker]) and call :meth:snapshot at any time — from the event loop or from another thread — to get a :class:ProgressSnapshot. The tracker is updated synchronously as each event is emitted (it does not go through the callback queue), so a snapshot is never behind the pipeline.

The state lives only in the memory of the current process: it is not persisted and is not visible from other processes. One instance is meant for one run_pipeline() call; a second pipeline_started event resets it.

snapshot

snapshot() -> ProgressSnapshot

Return the current progress of the run. Thread-safe.

The returned snapshot is an independent copy: later events do not change it.

scinr.newton.progress.ProgressSnapshot dataclass

Point-in-time view of a run, returned by :meth:ProgressTracker.snapshot.

Attributes:

Name	Type	Description
status	Literal['pending', 'running', 'success', 'completed_with_errors', 'failed']	"pending" (no event yet), "running", "success", "completed_with_errors" (the run returned with PipelineResult.success == False) or "failed" (the run raised).
job_id	str | None	The job_id passed to run_pipeline().
started_at	datetime | None	When the run started (UTC), or None while pending.
finished_at	datetime | None	When the run ended (UTC), or None while not finished.
elapsed_s	float	Seconds since the run started, up to finished_at once finished.
stages	tuple[str, ...]	Document stages that will run (empty until documents are discovered).
total_documents	int | None	Number of documents to process, or None until discovered.
documents_pending	int	Documents not started yet (0 while unknown).
documents_running	int	Documents in progress.
documents_succeeded	int	Documents finished with status "success".
documents_partial	int	Documents finished with status "partial".
documents_failed	int	Documents finished with status "failed".
documents_skipped	int	Documents finished with status "skipped".
running_by_stage	Mapping[str, int]	Number of documents currently in each stage.
completed_by_stage	Mapping[str, int]	Number of documents that completed each stage.
failed_by_stage	Mapping[str, int]	Number of documents that failed at each stage.
tabular	Literal['running', 'completed', 'completed_with_errors'] | None	State of the tabular batch: "running", "completed", "completed_with_errors", or None when no tabular files are processed.
percent	float	Overall document progress, 0-100, rounded to one decimal. The tabular batch does not count towards it.
documents	Mapping[int, DocumentProgress]	Progress of every started document, keyed by document_index.
error	str | None	Error message when the run raised, otherwise None.

scinr.newton.progress.DocumentProgress dataclass

Progress of one document, as seen by :class:ProgressTracker.

Attributes:

Name	Type	Description
document_index	int	Position of the document among the discovered documents.
doc_path	str | None	Path of the document.
document_name	str | None	Latest known document name.
tenant_id	str | None	Tenant of the document (None = public).
status	Literal['running', 'success', 'partial', 'failed', 'skipped']	"running" until the document finishes, then "success", "partial", "failed" or "skipped".
current_stage	str | None	Stage in progress right now, or None.
planned_stages	tuple[str, ...]	Stages this document will run.
completed_stages	tuple[str, ...]	Stages this document completed, in order.
stopped_at	str | None	Stage where the document stopped because of a failure, or None.
error	str | None	Error message when the document failed, or None.
started_at	datetime	When the document started (UTC).
finished_at	datetime | None	When the document finished (UTC), or None while running.