fastrag.ingestion.steps.fetching.sources#

Sources: aggregate several SourceSteps into one stream.

docs/specification/subclasses.md, resolved against the base class task 003 actually built (which exposes config(), not the pseudocode’s compute_config_hash() — see .agent/results/003-step-base.md’s own Deviations for why the pseudocode is stale here too).

Classes

Sources(sources[, cache, policy, filter, ...])

Aggregator step combining multiple Fetchers or SourceSteps into one stream.
class fastrag.ingestion.steps.fetching.sources.Sources(sources, cache=None, policy=None, *, filter=None, max_sources=None)[source]#

Bases: SourceStep

Aggregator step combining multiple Fetchers or SourceSteps into one stream.

sources = Sources(
    sources=[LocalFileFetcher(...), SitemapCrawler(...)],
    cache=cache,  # default cache for all sources
    policy=policy,  # default policy for all sources
)
Parameters:
config()[source]#

No identity-affecting configuration of its own.

Sources is never cached as a single bucket (see __init__) so this only exists to satisfy PipelineStep.config’s abstract contract; nothing consults it.

Return type:

dict[str, Any]

async aclose()[source]#

Release resources held by children that expose an async close hook.

Return type:

None

async evaluate(context)[source]#

Merge every child’s stream, each cached under its own resolution.

Each child is driven through run_source_step() — the same caching + lifecycle wrap every SourceStep gets — under a context whose cache/policy are resolved once here: a child with its own explicit cache/policy keeps them (resolution happens again, inside the child’s own wrap, against whatever this context carries); a child with neither inherits this aggregator’s.

Parameters:

context (ExecutionContext)

Return type:

AsyncIterator[Source]

property cache: CacheStore | Literal[False] | None#

Return the cache explicitly configured for this step.

property config_hash: str#

A stable hash of this step’s class name and config().

Two identically configured steps of the same class hash equal; changing any declared config value changes the hash; the hash is stable across processes. Used as part of every cache id this step’s evaluation reads or writes.

effective_policy(context)#

Resolve this step’s policy over the one inherited from context.

The step’s own explicitly-set fields take priority; everything else falls through to the pipeline (or enclosing router/aggregator) policy.

Parameters:

context (ExecutionContext)

Return type:

Policy

event_name: str = 'Fetching'#

The stage this step’s work belongs to, e.g. "Chunking".

Declared as a plain class attribute by a family base class, and inherited by every concrete step in that family:

class Chunker(DocumentStep, ABC):
    event_name = "Chunking"


class ChunkerRouter(RoutedStep[Document], Chunker):
    ...  # -> "Chunking", inherited from `Chunker`

A concrete step overrides it to opt out of its family’s label:

class ChunkerRouter(RoutedStep[Document], Chunker):
    event_name = "ChunkerRouter"  # -> "ChunkerRouter"

Resolution walks the MRO for the first class that declares it in its own class body (so a mixin listed first, e.g. RoutedStep, contributes nothing unless it declares one itself), and falls back to the class’s own name when nothing in the MRO does — a step family that never declares an event_name still labels each of its steps distinctly rather than collapsing them under one blank heading. That fallback is computed per instance rather than assigned onto the class: assigning it would make it look “declared” to every subclass, which would then inherit their base’s name instead of getting their own.

property policy: Policy | None#

Return the policy explicitly configured for this step.

async run()#

Execute this step independently and collect every item.

Builds a default ExecutionContext when none is installed (this step’s own policy, or a fresh Policy). Intended primarily for testing and standalone use — pipeline-wide event observation is Pipeline’s job.

Return type:

list[T]

property upstream: PipelineStep[Any] | None#

Return the upstream step bound via __call__, or None.

Read-only counterpart to __call__’s self._upstream assignment — lets external code (e.g. Pipeline, walking a graph from its root to classify each step by position) introspect composition without reaching into a private attribute.

name: str#