fastrag.ingestion.steps.fetching#

route_for(uri)[source]#

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

Parameters:: uri (str)
Return type:: Fetcher | None

canonical_uri(uri)[source]#

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

Parameters:: uri (str)
Return type:: str

async fetch(uri)[source]#

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]

routes_config()[source]#

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

Return type:: list[dict[str, Any]]

property cache: CacheStore | Literal[False] | None#: Return the cache explicitly configured for this step.

abstractmethod config()#

Return this step’s own configuration, for cache-identity hashing.

Every concrete step declares this explicitly — steps are plain classes, not dataclasses, so there is no automatic field introspection. A step that forgets it cannot be instantiated: config is abstract, so construction itself fails loudly rather than config_hash silently collapsing to a constant.

Return type:: Mapping[str, Any]

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

abstractmethod evaluate(context)#

Lazily evaluate the step within an execution context.

Parameters:: context (ExecutionContext) – Execution-scoped state and services.
Yields:: Values produced by the step.
Return type:: AsyncIterator[T]

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#

class fastrag.ingestion.steps.fetching.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:

url (str)
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)

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:

target (str)
context (ExecutionContext)

Return type:

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:

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:

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#

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

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

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

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

Read one file, yielding the single Source it produces.

Parameters:

target (str)
context (ExecutionContext)

Return type:

async evaluate(context)[source]#

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

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:

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:

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#

class fastrag.ingestion.steps.fetching.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:

target (str)
context (ExecutionContext)

Return type:

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:

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:

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#

class fastrag.ingestion.steps.fetching.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:

sources (Sequence[SourceStep])
cache (CacheStore | Literal[False] | None)
policy (Policy | None)
filter (Filter[Source] | None)
max_sources (int | None)

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#

class fastrag.ingestion.steps.fetching.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 (Sequence[str] | None)
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.

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:

target (str)
context (ExecutionContext)

Return type:

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:

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:

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#

class fastrag.ingestion.steps.fetching.YouTubeFetcher(*, url=None, audio_only=False, cache=None, policy=None, cache_validation_interval=None)[source]#

Bases: Fetcher

Download one YouTube video, yielding its media bytes as a Source.

Needs the “video” extra (yt-dlp).

Parameters:

url (str | None)
audio_only (bool)
cache (CacheStore | Literal[False] | None)
policy (Policy | None)
cache_validation_interval (float | None)

config()[source]#

Cache fingerprint: whether the audio track alone is downloaded.

Return type:: dict[str, Any]

canonical_uri(uri)[source]#

The watch?v= URL for any spelling of a video URL.

Parameters:: uri (str)
Return type:: str

async targets(context)[source]#

The one video this fetcher targets.

Parameters:: context (ExecutionContext)
Return type:: list[str]

async evaluate_target(target, context)[source]#

Download the video, yielding its Source.

Parameters:

target (str)
context (ExecutionContext)

Return type:

async evaluate(context)[source]#

Download the video, uncached.

Parameters:: context (ExecutionContext)
Return type:: AsyncIterator[Source]

async is_source_modified(source_id, cached_etag, cached_last_modified, cached_content_hash)[source]#

Never: a video’s bytes are keyed by its id and do not change.

Parameters:

source_id (str)
cached_etag (str | None)
cached_last_modified (datetime | None)
cached_content_hash (str)

Return type:

async fetch(uri)[source]#

Download the video at uri (any YouTube URL form).

Parameters:: uri (str)
Return type:: list[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.

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: