fastrag.ingestion.logger.console#

ConsoleLogger: renders a pipeline’s event stream to the console.

docs/specification/logging.md / docs/specification/API.md’s “Example output”: one row per leaf fetcher (LocalFileFetcher, WebCrawler, SitemapCrawler, S3Fetcher, …), one column per pipeline stage (Fetching, Parsing, Chunking, Embedding, Indexing, …), each cell new/cached/errors.

Two separate questions have to be answered to fill in a cell: which column (which stage produced this event) and which row (which fetcher did the item ultimately come from). They are solved differently, because the information each needs is only available in a different place:

  • Column is whatever the emitting step’s event_name resolves to, read straight off the event (every event carries it). A step family declares its own label (Chunker.event_name = "Chunking"), a step that overrides it is columned under its override, and a step whose family declares none falls back to its own class name — so a graph with a stage this SDK has never heard of grows a column for it rather than dropping its events. Columns are seeded in pipeline order by walking root’s upstream chain from the source back down to the root. Columns are then ordered by column_order, defaulting to Fetching, Parsing, Chunking, Embedding, Indexing. Unlisted labels follow in discovery order, including labels first seen on events from routed children. A step that emits no events at all still gets its (empty) column from that seeding.

  • Row is the name of the leaf SourceStep (fetcher) an item was originally produced by — knowable structurally only at the moment of fetching. fastrag.ingestion.steps.base._run_step stamps it there, into the reserved Source.metadata["source_step"] key, onto every Source that doesn’t already carry one (so nested composition — a Sources aggregate wrapping several fetchers — never overwrites the innermost, correct stamp with the aggregate’s own name). This module enumerates the known row names once at construction, by walking Sources.sources recursively (_leaf_fetcher_names) if the graph’s source-most step is a Sources aggregate, or taking that step’s own name otherwise.

    A Document five stages downstream never re-reads that metadata key directly (parsing/chunking/embedding never touch it, so a stale parse or chunk cache entry from before this reserved key existed would carry none of it). Instead, this module tracks a live Source.id -> row name map, populated as each DocumentFetched arrives, and looks a Document up by its own source_id field — which every downstream step already carries forward unchanged — so attribution survives regardless of which stage’s cache happens to be warm.

Rows are named after a step instance (ItemFailed.step, metadata["source_step"]) and columns after a step stage (event_name) — the two never collide, because one fetcher’s row and the Fetching column answer different questions about the same event.

new/cached are read directly off the item every domain event carries (DocumentFetched.source, DocumentParsed/DocumentChunked/ DocumentEmbedded/DocumentSaved’s .document) — every one of these events now fires the same way for a cache hit as for fresh work (see fastrag.ingestion.steps.base.DocumentStep._on_produced), so item.cached is trustworthy on every event this module sees. A DocumentFetched is counted at most once per Source.id: nested SourceStep composition emits it once per level (the leaf fetcher’s own, and again when the enclosing Sources aggregate is itself driven) — by design, the same way lifecycle events double up — so this module dedupes rather than double-counting the same fetch.

errors comes from ItemFailed, columned by its event_name like any other event. Its row is resolved from its step when that directly names a known leaf fetcher (a fetch-stage failure — a fetcher’s own run_concurrently call always names itself), and otherwise from its source_id through the same live map DocumentFetched populates (a parser/chunker/embedder/sink step is shared by every row, so only the item it was processing says which one). A failure this module cannot attribute to a row either way (an empty step and an unknown source_id) is not silently dropped: it is counted separately and surfaced as a trailing line once the run ends, so an attribution gap is visible rather than invisible.

A graph whose source-most step is not a SourceStep at all (i.e. cannot be introspected — some hand-built test double) falls back to the single-row behavior this module always had: every event lands in one row, keyed _UNATTRIBUTED_ROW, rather than crashing.

Functions

format_duration(seconds)

8.2s, 2m05s, 1h02m: short, fixed-ish width, human readable.

Classes

ConsoleLogger(pipeline, *[, show_progress, ...])

Renders a pipeline's lifecycle and domain events to the console.
fastrag.ingestion.logger.console.format_duration(seconds)[source]#

8.2s, 2m05s, 1h02m: short, fixed-ish width, human readable.

Parameters:

seconds (float)

Return type:

str

class fastrag.ingestion.logger.console.ConsoleLogger(pipeline, *, show_progress=True, show_domain_events=False, show_errors=False, refresh_interval=0.5, column_order=None, stream=None, heartbeat_interval=15.0, slow_after=60.0, stall_after=30.0)[source]#

Bases: Logger

Renders a pipeline’s lifecycle and domain events to the console.

Aggregates item-level events into the per-source, per-stage progress table docs/specification/API.md’s “Example output” shows — one row per leaf fetcher, one column per stage, each cell new/cached/errors — refreshed in place on a TTY, or printed once at the end otherwise (piping to a file must never see escape sequences).

Parameters:

  • pipeline (Pipeline)

  • show_progress (bool)

  • show_domain_events (bool)

  • show_errors (bool)

  • refresh_interval (float)

  • column_order (Sequence[str] | None)

  • stream (Any)

  • heartbeat_interval (float)

  • slow_after (float)

  • stall_after (float)

show_progress: bool#
show_domain_events: bool#
show_errors: bool#
refresh_interval: float#
async run()[source]#

Consume the pipeline’s event stream and render progress as it arrives.

Iterating self.pipeline both drives the root step to completion and yields each event; a step failure under policy.on_error == "raise" propagates out of this call the same way it does out of Pipeline.run(), after every event already produced has been handled. If any ItemFailed could not be attributed to a known row, a trailing summary line is written after the final render so that failure is never silently dropped.

A background ticker keeps the output alive while no event arrives: on a TTY it redraws the table and status every refresh_interval; otherwise it prints a status line every heartbeat_interval. Failures are printed as they happen when show_errors is set, and listed again at the end.

Return type:

None

pipeline: Pipeline#