windlass.interfaces.loader

loader

The document-loader interface.

A loader turns a source — a path, a glob, a URL, a byte stream — into :class:~windlass.core.types.Document objects. Format detection is automatic: :func:~windlass.rag.loading.load inspects the extension and MIME type and picks the right registered loader, so .ingest("./docs") handles a folder of mixed PDFs, spreadsheets and Markdown without any configuration.

Implementers override one coroutine, :meth:Loader.aload_source.

Example

import tempfile, pathlib p = pathlib.Path(tempfile.mkdtemp()) / "note.txt" _ = p.write_text("hello world") from windlass.providers.loaders.text import TextLoader TextLoader().load(p)[0].content 'hello world'

Loader

Loader(
    *,
    encoding: str = "utf-8",
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract document loader.

Parameters:

Name	Type	Description	Default
encoding	str	Text encoding used for character-based formats.	'utf-8'
metadata	dict[str, Any] | None	Extra metadata merged into every produced document — useful for tagging a whole corpus with a tenant or collection id.	None
recursive	bool	Whether directory sources are walked recursively.	True
on_error	str	"raise" aborts the batch, "skip" logs and continues. "skip" is the right default for a 10,000-file corpus where one corrupt PDF should not lose the run.	'skip'
name	str | None	Component name for traces.	None
**config	Any	Loader-specific options.	{}

Attributes:

Name	Type	Description
extensions	tuple[str, ...]	File suffixes this loader claims, used for auto-detection.
mimetypes	tuple[str, ...]	MIME types this loader claims.

Example

Implementing a loader takes one method::

class MyLoader(Loader):
    provider_name = "mine"
    extensions = (".mine",)

    async def aload_source(self, source):
        return [Document(content=read(source), source=str(source))]

Source code in src\windlass\interfaces\loader.py

def __init__(
    self,
    *,
    encoding: str = "utf-8",
    metadata: dict[str, Any] | None = None,
    recursive: bool = True,
    on_error: str = "skip",
    name: str | None = None,
    **config: Any,
) -> None:
    if on_error not in {"raise", "skip"}:
        raise ValueError("on_error must be 'raise' or 'skip'")
    super().__init__(
        name=name or self.provider_name,
        encoding=encoding,
        metadata=metadata or {},
        recursive=recursive,
        on_error=on_error,
        **config,
    )
    self.encoding = encoding
    self.extra_metadata: dict[str, Any] = dict(metadata or {})
    self.recursive = recursive
    self.on_error = on_error

aload_source abstractmethod async

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

Load one concrete source.

Called once per file. Directory expansion and error handling are done for you by :meth:aload.

Parameters:

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

Returns:

Type	Description
list[Document]	The documents extracted from it. A PDF may return one document per
list[Document]	page or one for the whole file — that is the loader's choice.

Raises:

Type	Description
IngestionError	When the source cannot be parsed.

Source code in src\windlass\interfaces\loader.py

@abc.abstractmethod
async def aload_source(self, source: SourceLike) -> list[Document]:
    """Load one concrete source.

    Called once per file. Directory expansion and error handling are done
    for you by :meth:`aload`.

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

    Returns:
        The documents extracted from it. A PDF may return one document per
        page or one for the whole file — that is the loader's choice.

    Raises:
        IngestionError: When the source cannot be parsed.
    """

can_handle classmethod

can_handle(source: SourceLike) -> bool

Return whether this loader claims source.

Parameters:

Name	Type	Description	Default
source	SourceLike	Path, URL or payload to test.	required

Returns:

Type	Description
bool	True when the extension or scheme matches this loader.

Example

from windlass.providers.loaders.text import TextLoader TextLoader.can_handle("notes.txt") True TextLoader.can_handle("scan.pdf") False

Source code in src\windlass\interfaces\loader.py

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return whether this loader claims ``source``.

    Args:
        source: Path, URL or payload to test.

    Returns:
        True when the extension or scheme matches this loader.

    Example:
        >>> from windlass.providers.loaders.text import TextLoader
        >>> TextLoader.can_handle("notes.txt")
        True
        >>> TextLoader.can_handle("scan.pdf")
        False
    """
    if isinstance(source, str | Path):
        text = str(source)
        if text.startswith(("http://", "https://")):
            return cls.handles_urls
        return Path(text).suffix.lower() in cls.extensions
    return False

aload async

aload(
    source: SourceLike | Sequence[SourceLike], *, concurrency: int | None = None
) -> list[Document]

Load one or many sources.

Directories are expanded (respecting :attr:recursive and this loader's :attr:extensions), and files are read concurrently.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, byte payload, or a sequence of them. A directory path expands to its matching files.	required
concurrency	int | None	Maximum simultaneous reads. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Document]	Every extracted document, with :attr:extra_metadata merged in.

Raises:

Type	Description
IngestionError	When on_error='raise' and a source fails, or when no source could be read at all.

Performance

Blocking parsers run on the thread pool, so a folder of 500 PDFs is parsed in parallel rather than serially.

Source code in src\windlass\interfaces\loader.py

async def aload(
    self, source: SourceLike | Sequence[SourceLike], *, concurrency: int | None = None
) -> list[Document]:
    """Load one or many sources.

    Directories are expanded (respecting :attr:`recursive` and this loader's
    :attr:`extensions`), and files are read concurrently.

    Args:
        source: A path, URL, byte payload, or a sequence of them. A
            directory path expands to its matching files.
        concurrency: Maximum simultaneous reads. Defaults to the global
            ``max_concurrency`` setting.

    Returns:
        Every extracted document, with :attr:`extra_metadata` merged in.

    Raises:
        IngestionError: When ``on_error='raise'`` and a source fails, or
            when no source could be read at all.

    Performance:
        Blocking parsers run on the thread pool, so a folder of 500 PDFs is
        parsed in parallel rather than serially.
    """
    sources = self._expand(source)
    if not sources:
        return []

    limit = concurrency or settings().max_concurrency
    results = await gather_bounded(
        [self._load_guarded(s) for s in sources], limit=limit, return_exceptions=True
    )

    documents: list[Document] = []
    failures: list[tuple[Any, BaseException]] = []
    for src, result in zip(sources, results, strict=True):
        if isinstance(result, BaseException):
            failures.append((src, result))
            continue
        documents.extend(result)

    if failures:
        if self.on_error == "raise":
            src, exc = failures[0]
            raise IngestionError(
                f"Failed to load {src}: {exc}",
                context={"source": str(src), "failures": len(failures)},
            ) from exc
        for src, exc in failures:
            self._log.warning("Skipping %s: %s", src, exc)
        if not documents:
            src, exc = failures[0]
            raise IngestionError(
                f"No documents could be loaded; the first failure was {src}: {exc}",
                hint="Check the paths and that the right extra is installed.",
                context={"failures": len(failures)},
            ) from exc

    for doc in documents:
        if self.extra_metadata:
            doc.metadata = {**self.extra_metadata, **doc.metadata}
        doc.metadata.setdefault("loader", self.name)
    return documents

load

load(source: SourceLike | Sequence[SourceLike]) -> list[Document]

Blocking :meth:aload.

Source code in src\windlass\interfaces\loader.py

def load(self, source: SourceLike | Sequence[SourceLike]) -> list[Document]:
    """Blocking :meth:`aload`."""
    return run_sync(self.aload(source))

astream_load async

astream_load(source: SourceLike | Sequence[SourceLike]) -> AsyncIterator[Document]

Yield documents one at a time instead of building a list.

Use this for corpora too large to hold in memory: combined with Pipeline.aingest_stream it keeps peak memory flat regardless of corpus size.

Parameters:

Name	Type	Description	Default
source	SourceLike | Sequence[SourceLike]	A path, URL, payload, or sequence of them.	required

Yields:

Type	Description
AsyncIterator[Document]	Documents as each source finishes parsing.

Source code in src\windlass\interfaces\loader.py

async def astream_load(
    self, source: SourceLike | Sequence[SourceLike]
) -> AsyncIterator[Document]:
    """Yield documents one at a time instead of building a list.

    Use this for corpora too large to hold in memory: combined with
    ``Pipeline.aingest_stream`` it keeps peak memory flat regardless of
    corpus size.

    Args:
        source: A path, URL, payload, or sequence of them.

    Yields:
        Documents as each source finishes parsing.
    """
    for item in self._expand(source):
        try:
            docs = await self.aload_source(item)
        except Exception as exc:
            if self.on_error == "raise":
                raise IngestionError(f"Failed to load {_label(item)}: {exc}") from exc
            self._log.warning("Skipping %s: %s", item, exc)
            continue
        for doc in docs:
            if self.extra_metadata:
                doc.metadata = {**self.extra_metadata, **doc.metadata}
            doc.metadata.setdefault("loader", self.name)
            yield doc