fastrag.ingestion.steps.fetching.youtube#

YouTube video fetching, and the iframe-embed discovery crawlers route through it.

YouTubeFetcher downloads one video (or just its audio track) with yt-dlp. It is a normal Fetcher usable on its own, and the target of route() on crawlers:

crawler = WebCrawler(url="https://example.com").route(
    youtube_filter, YouTubeFetcher(audio_only=True)
)

Module Attributes

youtube_filter

Matches any URL YouTubeFetcher can download.

Functions

embed_urls(content, base_uri)

Absolute src URLs of every <iframe> in content.

fetch_embeds(owner, page_uri, *contents)

Fetch the page's <iframe> embeds that owner has a route for.

youtube_video_id(uri)

The video id of a YouTube watch/embed/short/youtu.be URL, else None.

Classes

YouTubeFetcher(*[, url, audio_only, cache, ...])

Download one YouTube video, yielding its media bytes as a Source.
fastrag.ingestion.steps.fetching.youtube.youtube_filter = <fastrag.ingestion.filters.url.URLFilter object>#

Matches any URL YouTubeFetcher can download.

Parameters:

item (T_contra)

Return type:

bool

fastrag.ingestion.steps.fetching.youtube.youtube_video_id(uri)[source]#

The video id of a YouTube watch/embed/short/youtu.be URL, else None.

Parameters:

uri (str)

Return type:

str | None

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

AsyncIterator[Source]

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:

bool

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:

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#
fastrag.ingestion.steps.fetching.youtube.embed_urls(content, base_uri)[source]#

Absolute src URLs of every <iframe> in content.

Parameters:
Return type:

list[str]

async fastrag.ingestion.steps.fetching.youtube.fetch_embeds(owner, page_uri, *contents)[source]#

Fetch the page’s <iframe> embeds that owner has a route for.

contents are (bytes, base_uri) pairs — typically the raw and the rendered page, since a renderer may strip embeds. Equivalent spellings of one embed are fetched once. A dead or private embed is skipped with a warning rather than failing the page.

Parameters:
Return type:

list[Source]