fastrag.ingestion

fastrag.ingestion — lazy, event-emitting ingestion pipeline.

A lazy pipeline API: constructing a step never executes anything, and evaluating one emits lifecycle and domain events instead of writing to stdout directly. This module re-exports the inert value types owned by ingestion (Source, policy, events), the step contract and its evaluator (PipelineStep, ExecutionContext), and Pipeline, the context manager that installs the execution context and event sink for a whole graph. Document — shared by retrieval, querying, serving and benchmarking, not an ingestion-only concern — lives at fastrag.models instead and is deliberately not re-exported here. The cache contract lives at fastrag.ingestion.cache and is deliberately not re-exported here (see its own module docstring); Logger/ConsoleLogger live at fastrag.ingestion.logger, likewise not re-exported here. This package must never import from outside its own tree, except the foundation modules every package sits on top of (fastrag.models, fastrag.exceptions) and the underscore-prefixed private modules directly under fastrag (fastrag._sql today) that hold plumbing shared across packages.

class fastrag.ingestion.Pipeline(root, *, cache=None, policy=None, source_headers_only=False, headers_only=False)

Bases: AbstractAsyncContextManager[Pipeline]

Evaluates a lazy PipelineStep graph.

Pipeline owns the ExecutionContext for a whole run: it is responsible for creating it (with this pipeline’s own default cache/policy) and for installing the event sink every step in the graph — however deeply nested — ultimately emits into.

async with Pipeline(vector_store.save(embedded_documents)) as pipeline:
    await ConsoleLogger(pipeline).run()

Parameters:

• root (PipelineStep[Any])

• cache (CacheStore | None)

• policy (Policy | None)

• source_headers_only (bool)

• headers_only (bool)

property root: PipelineStep[Any]

The step this pipeline evaluates — and the top of its upstream chain.

async run()

Evaluate root while discarding every event it produces.

For when event consumption is handled by something else — a Logger iterating this same pipeline concurrently, e.g. — rather than by the caller of run() itself.

Return type:

None

class fastrag.ingestion.Policy(retries=UNSET, backoff=UNSET, max_backoff=UNSET, timeout=UNSET, on_error=UNSET, concurrency=UNSET, batch_size=UNSET, cache_batch_size=UNSET, recreate_cache=UNSET)

Bases: object

Execution policy controlling operational constraints for pipeline steps.

Parameters:

• retries (int)

• backoff (float)

• max_backoff (float)

• timeout (float)

• on_error (Literal['skip', 'raise', 'retry'])

• concurrency (int)

• batch_size (int)

• cache_batch_size (int)

• recreate_cache (bool)

retries

Maximum number of execution retries on failure.

Type:

int

backoff

Initial exponential backoff delay in seconds.

Type:

float

max_backoff

Ceiling delay for exponential backoff in seconds.

Type:

float

timeout

Maximum execution duration per item or batch in seconds.

Type:

float

on_error

Failure handling strategy ("skip", "raise", "retry").

Type:

Literal[‘skip’, ‘raise’, ‘retry’]

concurrency

Max concurrent worker tasks.

Type:

int

batch_size

Processing batch size for batched transforms.

Type:

int

cache_batch_size

Cache lookup window, independent of transform concurrency.

Type:

int

recreate_cache

Skip cached results and overwrite them with freshly recomputed values.

Type:

bool

A step’s policy is layered over its parent’s (the pipeline’s, or an enclosing router/aggregator’s) with merge(): only the fields the caller explicitly passed override the parent; the rest are inherited.

merge(base)

Return this policy layered over base.

Fields this policy set explicitly win; every other field is taken from base. The result’s explicit set is the union of both, so a chain of merges keeps resolving correctly.

Parameters:

base (Policy | None) – The inherited (parent) policy, or None for no parent.

Return type:

Policy

delay_for(attempt)

Return the capped exponential backoff delay before attempt.

Parameters:

attempt (int) – The retry attempt number, 1 for the first retry.

Returns:

backoff * 2 ** (attempt - 1), capped at max_backoff.

Return type:

float

class fastrag.ingestion.Source(id, uri, content, content_type, content_hash, etag=None, last_modified=None, metadata=<factory>, last_validated=None, cached=False)

Bases: object

Raw content fetched from somewhere, plus enough metadata to place it.

A Source is produced by a fetcher and carries the original bytes together with enough information to identify, validate, cache, and later transform it into one or more Document instances. Sources are independently cacheable, and their freshness may be validated against the originating system using ETags, modification timestamps, content hashes, or equivalent mechanisms.

Parameters:

• id (str)

• uri (str)

• content (bytes)

• content_type (str)

• content_hash (str)

• etag (str | None)

• last_modified (datetime | None)

• metadata (dict[str, Any])

• last_validated (datetime | None)

• cached (bool)

id

Unique identifier for this source.

Type:

str

uri

Where the content came from.

Type:

str

content

The raw fetched bytes.

Type:

bytes

content_type

The source’s MIME media type.

Type:

str

content_hash

Deterministic hash of content, used as its address.

Type:

str

etag

HTTP validator for conditional refetching, when the origin supports it.

Type:

str | None

last_modified

Origin-reported modification time, when available.

Type:

datetime.datetime | None

metadata

Extended, source-specific metadata.

Type:

dict[str, Any]

last_validated

When this source was last confirmed fresh against its origin (or, for one that has never been revalidated, first fetched). Persisted: round-tripped through the cache so that SourceStep.cache_validation_interval means the same thing across process restarts, not just within one. None on a Source that has never been through a cache write, e.g. one just produced by evaluate() and not yet cached.

Type:

datetime.datetime | None

cached

Whether this instance was served from the cache rather than freshly fetched. Runtime-only: never persisted.

Type:

bool

Modules

cache	The pluggable step cache.
events	Events emitted during pipeline execution.
filters	fastrag.ingestion.filters — composable predicates over sources and documents.
logger	Pipeline event consumers for progress, documents and raw sources.
models	Value types produced by the ingestion pipeline's source-fetching stage.
pipeline	Pipeline: the context manager that evaluates a lazy step graph.
policy	Execution policy controlling operational constraints for pipeline steps.
steps	Step base classes and the policy-aware evaluator.