fastrag.ingestion.steps.fetching.web#

Link-following crawling: start from one or more URLs, follow what they link to.

Unlike SitemapCrawler (which trusts a sitemap to enumerate every page up front), WebCrawler discovers pages by fetching a start URL, extracting its links, and following them within the configured bounds — the right tool for a site with no sitemap, or where only a subtree reachable from one URL matters.

Caching is bucketed per start URL, not per page: targets() returns the (normalized) start URLs, and evaluate_target() crawls the whole subtree reachable from one of them, exactly docs/specification/API.md’s “Example 1” (source_cache_id = "webcrawler:1a2b3c4" -> [Source(a.com/1), Source(a.com/2), ...]). This means a start URL whose linked pages changed at all is revalidated (and, if stale, recrawled) as one unit — the granularity API.md’s own example describes, not a per-page bucket the way SitemapCrawler uses. See .agent/results/013-web-crawler.md for why this differs from SitemapCrawler’s per-page bucketing despite both opting into the same targets/evaluate_target hooks.

Classes

WebCrawler(*[, url, start_urls, max_urls, ...])

Follow links from one or more start URLs, within configured bounds.
class fastrag.ingestion.steps.fetching.web.WebCrawler(*, url=None, start_urls=None, max_urls=None, max_depth=None, include=None, exclude=None, same_origin=True, 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

Follow links from one or more start URLs, within configured bounds.

Accepts either a single url= (the form examples/08_sources.py/09_filters.py use) or several start_urls= (docs/specification/API.md’s multi-start form) — exactly one of the two must be given. Both are stored the same way internally (self.start_urls), so the rest of this class does not distinguish them.

from fastrag.ingestion.steps.fetching import WebCrawler

crawler = WebCrawler(url="https://docs.pontus-x.eu/", max_urls=200)
crawler = WebCrawler(start_urls=["a.com", "https://b.com/"], max_depth=3)
Parameters:

  • url (str | None)

  • start_urls (list[str])

  • max_urls (int | None)

  • max_depth (int | None)

  • include (str | Sequence[str] | None)

  • exclude (str | Sequence[str] | None)

  • same_origin (bool)

  • renderer (Renderer | None)

  • headers (Mapping[str, str] | None)

  • timeout (float)

  • follow_redirects (bool)

  • user_agent (str)

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

  • policy (Policy | None)

  • cache_validation_interval (float | None)

streams_targets = True#

A start URL’s crawl bounds, retries and times out each page itself, so the cache wrap streams it rather than running it as one timed item.

start_urls: list[str]#
config()[source]#

This crawler’s cache fingerprint: only what changes a page’s bytes.

start_urls, max_urls, max_depth, include, exclude, and same_origin decide which pages are discovered, not what a given page returns, so they are deliberately excluded here — the same principle config() follows, for the same reason. Per-start-URL cache ids already encode the start URL itself as their target.

Return type:

dict[str, Any]

async targets(context)[source]#

Return the normalized start URLs — one independent cache bucket each.

Parameters:

context (ExecutionContext)

Return type:

list[str]

async evaluate_target(target, context)[source]#

Crawl the subtree reachable from one start URL, within bounds.

Fetches breadth-first, one level of link depth at a time, each level’s pages fetched concurrently (bounded by policy.concurrency) via run_concurrently() — so a page that fails to fetch after retries emits an attributed ItemFailed and is skipped (policy.on_error == "skip") rather than aborting the whole crawl. A visited set, seeded with target itself, guarantees every URL is fetched at most once per call even when several pages link to it, and pages are deduplicated again by the Source.id they actually resolve to, so two URLs that redirect to one page are yielded (and cached) once.

Parameters:
Return type:

AsyncIterator[Source]

async evaluate(context)[source]#

Crawl every start URL, 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. Start URLs are themselves fetched concurrently, mirroring evaluate_target()’s own per-level concurrency.

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(), applied here to one page within a start URL’s cached bucket — the base class’s _any_modified calls this once per cached page, and treats the whole bucket as stale (recrawling the entire start URL via evaluate_target()) if any one page reports modified.

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 and renderer resources. Idempotent.

Return type:

None

keeps_source(source)[source]#

Whether source passes include/exclude (applied to cached output too).

Parameters:

source (Source)

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#