windlass.rag.loading

loading

Automatic loader selection.

rag.ingest("./docs") on a folder of mixed PDFs, spreadsheets, Markdown and URLs should just work. That is this module's job: inspect each source, pick the registered loader that claims it, and dispatch.

Selection order:

1. A URL goes to the YouTube loader if it looks like a video, otherwise the web loader.
2. A file goes to whichever registered loader claims its extension.
3. An unclaimed text-like file falls back to the plain-text loader.
4. Anything else raises with a list of what is supported, so the fix is obvious (usually pip install "windlass[loaders]").

Example

loader_for("notes.md").provider_name 'markdown' loader_for("https://example.com").provider_name 'web'

AutoLoader

AutoLoader(
    *,
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    loader_config: dict[str, dict[str, Any]] | None = None,
    **config: Any
)

Bases: Loader

Dispatches each source to the loader that claims it.

This is what makes ingest('./docs') handle a folder of mixed formats. Files with no matching loader are skipped with a warning rather than aborting the run, so one unreadable file cannot lose a large ingestion.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any] | None	Metadata attached to every produced document.	None
recursive	bool	Walk directories recursively.	True
on_error	str	"skip" (default) or "raise".	'skip'
loader_config	dict[str, dict[str, Any]] | None	Per-loader keyword arguments, keyed by loader name, e.g. {"pdf": {"per_page": False}}.	None
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib folder = pathlib.Path(tempfile.mkdtemp()) _ = (folder / "a.txt").write_text("plain") _ = (folder / "b.md").write_text("# heading") sorted(d.metadata["extension"] for d in AutoLoader().load(folder)) ['.md', '.txt']

Source code in src\windlass\rag\loading.py

def __init__(
    self,
    *,
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    loader_config: dict[str, dict[str, Any]] | None = None,
    **config: Any,
) -> None:
    super().__init__(metadata=metadata, recursive=recursive, on_error=on_error, **config)
    self.loader_config = dict(loader_config or {})
    self._cache: dict[str, Loader] = {}

can_handle classmethod

can_handle(source: SourceLike) -> bool

Return True — the auto loader accepts anything and routes it.

Source code in src\windlass\rag\loading.py

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return True — the auto loader accepts anything and routes it."""
    return True

aload_source async

aload_source(source: SourceLike) -> list[Document]

Route one source to its loader and read it.

Parameters:

Name	Type	Description	Default
source	SourceLike	A single path, URL or byte payload.	required

Returns:

Type	Description
list[Document]	The extracted documents, or [] when no loader claims the source
list[Document]	and on_error='skip'.

Raises:

Type	Description
IngestionError	When no loader claims the source and on_error='raise'.

Source code in src\windlass\rag\loading.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Route one source to its loader and read it.

    Args:
        source: A single path, URL or byte payload.

    Returns:
        The extracted documents, or ``[]`` when no loader claims the source
        and ``on_error='skip'``.

    Raises:
        IngestionError: When no loader claims the source and
            ``on_error='raise'``.
    """
    try:
        delegate = self._delegate(source)
    except IngestionError:
        if self.on_error == "raise":
            raise
        self._log.debug("No loader for %s; skipping.", source)
        return []
    return await delegate.aload_source(source)

extension_map

extension_map() -> dict[str, str]

Return the extension-to-loader-name mapping.

Built by asking every registered loader which extensions it claims, so a third-party loader registered via a plugin participates automatically.

Returns:

Type	Description
dict[str, str]	A {".pdf": "pdf", ...} mapping.

Example

extension_map()[".md"] 'markdown'

Source code in src\windlass\rag\loading.py

def extension_map() -> dict[str, str]:
    """Return the extension-to-loader-name mapping.

    Built by asking every registered loader which extensions it claims, so a
    third-party loader registered via a plugin participates automatically.

    Returns:
        A ``{".pdf": "pdf", ...}`` mapping.

    Example:
        >>> extension_map()[".md"]
        'markdown'
    """
    mapping: dict[str, str] = {}
    for spec in REGISTRY.specs("loader"):
        try:
            target = spec.resolve()
        except Exception:
            continue
        for extension in getattr(target, "extensions", ()):
            mapping.setdefault(extension, spec.name)
    return mapping

loader_for

loader_for(
    source: SourceLike | Sequence[SourceLike],
    *,
    metadata: dict[str, Any] | None = None,
    **config: Any
) -> Loader

Return a loader able to read source.

A heterogeneous sequence (or a directory containing several formats) yields an :class:AutoLoader, which dispatches per file.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, directory, byte payload, or sequence of them.	required
metadata	dict[str, Any] | None	Metadata attached to every produced document.	None
**config	Any	Forwarded to the chosen loader's constructor.	{}

Returns:

Type	Description
Loader	A ready-to-use loader.

Raises:

Type	Description
IngestionError	When nothing can read the source.

Example

loader_for("report.pdf").provider_name 'pdf' loader_for(b"raw bytes").provider_name 'text'

Source code in src\windlass\rag\loading.py

def loader_for(
    source: SourceLike | Sequence[SourceLike],
    *,
    metadata: dict[str, Any] | None = None,
    **config: Any,
) -> Loader:
    """Return a loader able to read ``source``.

    A heterogeneous sequence (or a directory containing several formats) yields
    an :class:`AutoLoader`, which dispatches per file.

    Args:
        source: A path, URL, directory, byte payload, or sequence of them.
        metadata: Metadata attached to every produced document.
        **config: Forwarded to the chosen loader's constructor.

    Returns:
        A ready-to-use loader.

    Raises:
        IngestionError: When nothing can read the source.

    Example:
        >>> loader_for("report.pdf").provider_name
        'pdf'
        >>> loader_for(b"raw bytes").provider_name
        'text'
    """
    if isinstance(source, Sequence) and not isinstance(source, str | bytes | Path):
        return AutoLoader(metadata=metadata, **config)

    if isinstance(source, bytes):
        return REGISTRY.create("loader", "text", metadata=metadata, **config)

    if isinstance(source, str) and source.startswith(("http://", "https://")):
        name = "youtube" if _is_youtube(source) else "web"
        return REGISTRY.create("loader", name, metadata=metadata, **config)

    path = Path(str(source))
    if path.is_dir() or any(ch in str(path) for ch in "*?["):
        return AutoLoader(metadata=metadata, **config)

    suffix = path.suffix.lower()
    claimed = extension_map().get(suffix)
    if claimed:
        return REGISTRY.create("loader", claimed, metadata=metadata, **config)

    if suffix in _TEXT_FALLBACK:
        return REGISTRY.create("loader", "text", metadata=metadata, **config)

    supported = ", ".join(sorted(extension_map())) or "(no loaders registered)"
    raise IngestionError(
        f"No loader can read {suffix or 'that source'!r}.",
        hint=f"Supported extensions: {supported}\n"
        'Most formats need: pip install "windlass[loaders]"',
        context={"source": str(source), "extension": suffix},
    )