fastrag.ingestion.steps.fetching.sitemap#

Sitemap-driven fetching: expand sitemap.xml, then fetch what it lists.

Ported from src/fastrag/steps/fetching/sitemap.py onto the new Fetcher/SourceStep base. Caching is per page (one bucket per crawled URL, via the targets/evaluate_target hooks fastrag.ingestion.steps.base fans out over) rather than per crawl, so a sitemap that gains one URL still serves every previously-fetched page from the cache — see docs/specification/API.md’s “cache each page independently to increase cache hit rate.”

Module Attributes

MAX_INDEX_DEPTH

How many nested <sitemapindex> levels to follow before giving up.

Classes

SitemapCrawler(*, url[, include, exclude, ...])

Expand sitemap.xml, then fetch what it lists.
fastrag.ingestion.steps.fetching.sitemap.MAX_INDEX_DEPTH = 3#

How many nested <sitemapindex> levels to follow before giving up. Real sites use one; a deeper chain is far more likely to be a loop.

class fastrag.ingestion.steps.fetching.sitemap.SitemapCrawler(*, url, include=None, exclude=None, max_urls=None, renderer=None, extract_pdfs=True, 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

Expand sitemap.xml, then fetch what it lists.

The right default for a website: the site itself publishes the canonical page list, so a rerun picks up new pages without editing any code. Nested <sitemapindex> documents are followed.

from fastrag.ingestion.steps.fetching import SitemapCrawler

crawler = SitemapCrawler(
    url="https://example.com/sitemap.xml",
    include=["https://example.com/docs/*"],
    cache=cache,
)
Parameters:

  • url (str)

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

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

  • max_urls (int | None)

  • renderer (Renderer | None)

  • extract_pdfs (bool)

  • 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)

config()[source]#

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

url, include, exclude, and max_urls decide which pages are discovered, not what a given page returns, so they are deliberately excluded here — otherwise raising max_urls would invalidate every already-cached page’s bucket for no reason. Per-page cache ids already encode the page’s own URL as their target.

Return type:

dict[str, Any]

async targets(context)[source]#

Return the filtered page URLs the sitemap lists, normalized and deduplicated.

Costs one request per sitemap document, and no page downloads. Called by the base class’s per-target cache wrap before deciding which pages need evaluate_target().

Sitemaps routinely list one page under several spellings (/audit and /audit/, a stray #fragment, an explicit :443). Those all collapse to one normalize_uri(), and so to one Source.id, so they are collapsed here too — into a single target and a single cache bucket, rather than fetched repeatedly and written over each other. include/exclude still see the sitemap’s own spelling for glob patterns; exact URLs match normalized spellings too.

Parameters:

context (ExecutionContext)

Return type:

list[str]

async evaluate_target(target, context)[source]#

Fetch one page URL, yielding its Source then one per routed embed.

Parameters:
Return type:

AsyncIterator[Source]

async evaluate(context)[source]#

Fetch every listed page, 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)[source]#

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

source_id is the cached entry’s normalized id, which — unlike API.md’s illustrative example — still carries 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 and renderer resources. 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#