fastrag.ingestion.steps.fetching.local

Local-disk fetching: files matched by glob, one Source per file.

Where FileSource fetches one URL and the crawlers discover pages over HTTP, LocalFileFetcher reads whatever a corpus that already lives on disk needs — “every spreadsheet under data/” — without publishing it over HTTP first. A pattern is a plain file path, a directory path (expanded to the files under it, recursively), or a glob (including **); several patterns may be given at once and their results are deduplicated.

Caching is one bucket per file, via the same targets/evaluate_target hooks SitemapCrawler uses for its per-page buckets — one new file appearing under a watched directory never invalidates the files already cached. Freshness is checked cheaply, without rereading a file’s bytes: a file’s size doubles as its “ETag” (an opaque validator, same slot HTTP’s ETag occupies) and its mtime as its “Last-Modified” — both available from a single stat() call. Content is only reread when either changes.

Classes

LocalFileFetcher(*, patterns[, include, ...])

Fetch local files matched by glob, yielding one Source per file.

class fastrag.ingestion.steps.fetching.local.LocalFileFetcher(*, patterns, include=None, exclude=None, max_files=None, cache=None, policy=None, cache_validation_interval=None)

Bases: Fetcher

Fetch local files matched by glob, yielding one Source per file.

from fastrag.ingestion.steps.fetching import LocalFileFetcher

fetcher = LocalFileFetcher(
    patterns=["data/**/*.xlsx"],
    exclude=["data/**/~$*"],
    cache=cache,
)

Each element of patterns may be a plain file path, a directory path (every file under it, recursively), or a glob pattern (*, ?, [...], and ** when used with a directory-spanning segment — see glob.glob()). A pattern that matches no files is an error, not a silent no-op: a typo’d path should fail loudly rather than quietly ingest nothing.

Parameters:

• patterns (Sequence[str])

• include (Sequence[str] | None)

• exclude (Sequence[str] | None)

• max_files (int | None)

• cache (CacheStore | Literal[False] | None)

• policy (Policy | None)

• cache_validation_interval (float | None)

config()

This fetcher’s cache fingerprint.

Empty: unlike the HTTP-speaking fetchers, nothing here changes what bytes a given file returns — that is entirely up to the filesystem. patterns/include/exclude/max_files decide which files are discovered, not what a given file’s bytes are, the same reasoning config() documents for its own URL selection — so they are deliberately excluded here too. Per-file cache ids already encode the file’s own URI as their target.

Return type:

dict[str, Any]

async targets(context)

Return the filtered, deduplicated file:// URIs this fetcher targets.

Called by the base class’s per-target cache wrap before deciding which files need evaluate_target(). Expansion (globbing, directory walks, and the stat calls needed to tell a file from a directory) runs in a worker thread so it never blocks the event loop.

Raises:

fastrag.exceptions.FetchError – A pattern in patterns matched no files.

Parameters:

context (ExecutionContext)

Return type:

list[str]

async evaluate_target(target, context)

Read one file, yielding the single Source it produces.

Parameters:

• target (str)

• context (ExecutionContext)

Return type:

AsyncIterator[Source]

async evaluate(context)

Read every matched file, uncached, bounded by policy.concurrency.

Used directly when no cache is in effect — the per-target cache wrap (targets()/evaluate_target()) is what applies when one is.

Parameters:

context (ExecutionContext)

Return type:

AsyncIterator[Source]

async is_source_modified(source_id, cached_etag, cached_last_modified, cached_content_hash)

Conditional freshness check: file size, then mtime — never rereading bytes.

cached_etag holds the file’s size at the time it was cached (this fetcher’s stand-in for an HTTP ETag, an opaque origin validator, populated the same way by _fetch_source) and cached_last_modified its mtime. A file that no longer exists, or can no longer be stat-ed, reads as modified — the same “unanswerable freshness question” logic is_source_modified() applies to a server that refuses HEAD: an attempted refetch always produces a definite answer (here, a FetchError), rather than silently keeping stale bytes.

Parameters:

• source_id (str)

• cached_etag (str | None)

• cached_last_modified (datetime | None)

• cached_content_hash (str)

Return type:

bool

property cache: CacheStore | Literal[False] | None

Return the cache explicitly configured for this step.

canonical_uri(uri)

Identity under which uri is fetched, to deduplicate equivalent spellings.

Parameters:

uri (str)

Return type:

str

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.

async fetch(uri)

Fetch one URL in-process, uncached. Needed to be a route() target.

Raises:

fastrag.exceptions.ConfigurationError – This fetcher cannot fetch an arbitrary URL.

Parameters:

uri (str)

Return type:

list[Source]

property policy: Policy | None

Return the policy explicitly configured for this step.

route(filter_rule, fetcher)

Delegate the URLs matching filter_rule to fetcher.

The filter sees a content-less Source carrying only the URL, so URL-based filters such as URLFilter apply. Routes are tried in registration order, and the first match wins.

Parameters:

• filter_rule (Filter[Source]) – Predicate over the URL about to be fetched.

• fetcher (Fetcher) – Fetcher implementing fetch(), used on match.

Returns:

Self, for fluent chaining.

Return type:

Self

route_for(uri)

The fetcher routed to uri, or None to fetch it normally.

Parameters:

uri (str)

Return type:

Fetcher | None

routes_config()

Routed fetchers’ fingerprints, for inclusion in a crawler’s config().

Return type:

list[dict[str, Any]]

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.

routes: list[tuple[Filter[Source], Fetcher]]

name: str