fastrag.ingestion.steps.fetching.file#

Single-URL fetching: exactly one file at exactly one URL.

Where SitemapCrawler expands a sitemap and WebCrawler follows links — both discovering many pages — FileSource expresses the simplest possible intent: “ingest this one file, at this one URL.” The right tool for a single published PDF, spreadsheet, or report, and for anything that would otherwise need a one-entry sitemap. Aggregating several FileSource\ s (or mixing one into a crawl) is Sources’s job, not this class’s — it deliberately takes one url, not a list.

Caching is one bucket for the URL, via the same targets/evaluate_target hooks SitemapCrawler uses for its per-page buckets — here there is only ever one target, but the shape (and so cache_validation_interval’s meaning) is identical.

Classes

FileSource(*, url[, renderer, headers, ...])

Fetch exactly one URL, yielding exactly one Source.
class fastrag.ingestion.steps.fetching.file.FileSource(*, url, renderer=None, headers=None, timeout=30.0, follow_redirects=True, user_agent='fastrag/0.1 (+https://github.com/AgrospAI/fastrag)', cache=None, policy=None, cache_validation_interval=None)[source]#

Bases: Fetcher

Fetch exactly one URL, yielding exactly one Source.

from fastrag.ingestion.steps.fetching import FileSource

source = FileSource(
    url="https://example.com/report.xlsx",
    cache=cache,
)
Parameters:
config()[source]#

This fetcher’s cache fingerprint: only what changes the fetched bytes.

url is deliberately excluded, exactly as config() excludes the pages it discovers — the URL is already encoded in the per-target cache id (see targets()), so including it here too would only make the fingerprint redundant, not more correct.

Return type:

dict[str, Any]

async targets(context)[source]#

Return the one normalized URL this fetcher targets.

Called by the base class’s per-target cache wrap before deciding whether evaluate_target() needs to run.

Parameters:

context (ExecutionContext)

Return type:

list[str]

async evaluate_target(target, context)[source]#

Fetch the URL, yielding the single Source it produces.

Parameters:
Return type:

AsyncIterator[Source]

async evaluate(context)[source]#

Fetch the URL, uncached.

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)[source]#

Conditional freshness check: ETag, then Last-Modified, then content hash.

Identical ladder to is_source_modified(): source_id is the cached entry’s normalized id, still carrying a scheme, so it is directly usable as a fetchable URI here. A server that answers none of the three (refuses HEAD and gives no reusable validator) reads as modified only implicitly: the refetch-and-compare fallback below always produces a definite answer.

Parameters:

  • source_id (str)

  • cached_etag (str | None)

  • cached_last_modified (datetime | None)

  • cached_content_hash (str)

Return type:

bool

async aclose()[source]#

Release the shared HTTP client. Idempotent.

Return type:

None

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#