windlass.providers.loaders.text

text

Plain-text, Markdown, JSON and CSV loaders — all dependency-free.

These four cover a surprising share of real corpora and they need nothing beyond the standard library, so pip install windlass alone can ingest a folder of notes, an exported dataset or a JSONL dump.

Example

import tempfile, pathlib d = pathlib.Path(tempfile.mkdtemp()) _ = (d / "note.md").write_text("# Title\n\nBody text.") docs = MarkdownLoader().load(d) docs[0].metadata["title"] 'Title'

TextLoader

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

Bases: Loader

Loads plain-text files.

Parameters:

Name	Type	Description	Default
encoding	str	Text encoding. Falls back to a lenient decode when the file turns out not to be valid in this encoding, so one mislabelled file does not abort a corpus.	'utf-8'
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib p = pathlib.Path(tempfile.mkdtemp()) / "a.txt" _ = p.write_text("hello") TextLoader().load(p)[0].content 'hello'

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 async

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

Read one text file.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path or a bytes payload.	required

Returns:

Type	Description
list[Document]	A single-element list holding the file's text.

Raises:

Type	Description
IngestionError	When the file cannot be read.

Source code in src\windlass\providers\loaders\text.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Read one text file.

    Args:
        source: A path or a bytes payload.

    Returns:
        A single-element list holding the file's text.

    Raises:
        IngestionError: When the file cannot be read.
    """
    raw = self._read_bytes(source)
    text = _decode(raw, self.encoding)
    metadata = self._base_metadata(source) if not isinstance(source, bytes) else {}
    return [
        Document(
            content=text,
            metadata={**metadata, "lines": text.count("\n") + 1},
            source=str(source) if not isinstance(source, bytes) else None,
            mimetype="text/plain",
        )
    ]

MarkdownLoader

MarkdownLoader(*, strip_frontmatter: bool = True, **config: Any)

Bases: Loader

Loads Markdown files and extracts their structure.

Pulls out the YAML front-matter (parsed as key/value pairs without needing PyYAML), the first # heading as a title, and the full heading list — all of which become metadata you can filter and cite on.

Parameters:

Name	Type	Description	Default
strip_frontmatter	bool	Remove the front-matter block from the content.	True
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib p = pathlib.Path(tempfile.mkdtemp()) / "d.md" _ = p.write_text("---\ntag: guide\n---\n# Setup\n\nRun it.") doc = MarkdownLoader().load(p)[0] doc.metadata["tag"], doc.metadata["title"] ('guide', 'Setup')

Source code in src\windlass\providers\loaders\text.py

def __init__(self, *, strip_frontmatter: bool = True, **config: Any) -> None:
    super().__init__(**config)
    self.strip_frontmatter = strip_frontmatter

aload_source async

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

Read one Markdown file.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path or a bytes payload.	required

Returns:

Type	Description
list[Document]	A single-element list.

Source code in src\windlass\providers\loaders\text.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Read one Markdown file.

    Args:
        source: A path or a bytes payload.

    Returns:
        A single-element list.
    """
    raw = self._read_bytes(source)
    text = _decode(raw, self.encoding)
    metadata: dict[str, Any] = (
        self._base_metadata(source) if not isinstance(source, bytes) else {}
    )

    match = _FRONTMATTER_RE.match(text)
    if match:
        metadata.update(_parse_frontmatter(match.group(1)))
        if self.strip_frontmatter:
            text = text[match.end() :]

    title = _TITLE_RE.search(text)
    if title:
        metadata.setdefault("title", title.group(1).strip())
    headings = [h.strip() for h in re.findall(r"^#{1,6}\s+(.+)$", text, re.MULTILINE)]
    if headings:
        metadata["headings"] = headings[:50]

    return [
        Document(
            content=text.strip(),
            metadata=metadata,
            source=str(source) if not isinstance(source, bytes) else None,
            mimetype="text/markdown",
        )
    ]

JSONLoader

JSONLoader(
    *,
    content_key: str | None = None,
    metadata_keys: list[str] | None = None,
    jq_path: str | None = None,
    **config: Any
)

Bases: Loader

Loads JSON and JSON Lines files.

Parameters:

Name	Type	Description	Default
content_key	str | None	Field to use as document content. When omitted the whole record is pretty-printed, which is right for heterogeneous data.	None
metadata_keys	list[str] | None	Fields copied into metadata. None copies every scalar field, which is usually what you want for filtering.	None
jq_path	str | None	Dotted path into the document to find the array of records, e.g. "data.items". Supports [] for list traversal.	None
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib, json p = pathlib.Path(tempfile.mkdtemp()) / "d.json" _ = p.write_text(json.dumps([{"text": "a", "id": 1}, {"text": "b", "id": 2}])) docs = JSONLoader(content_key="text").load(p) [d.content for d in docs]['a', 'b']

Source code in src\windlass\providers\loaders\text.py

def __init__(
    self,
    *,
    content_key: str | None = None,
    metadata_keys: list[str] | None = None,
    jq_path: str | None = None,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.content_key = content_key
    self.metadata_keys = metadata_keys
    self.jq_path = jq_path

aload_source async

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

Read a JSON or JSONL file.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path or a bytes payload.	required

Returns:

Type	Description
list[Document]	One document per record.

Raises:

Type	Description
IngestionError	When the file is not valid JSON.

Source code in src\windlass\providers\loaders\text.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Read a JSON or JSONL file.

    Args:
        source: A path or a bytes payload.

    Returns:
        One document per record.

    Raises:
        IngestionError: When the file is not valid JSON.
    """
    raw = self._read_bytes(source)
    text = _decode(raw, self.encoding)
    path = str(source) if not isinstance(source, bytes) else None
    base = self._base_metadata(source) if not isinstance(source, bytes) else {}

    records = self._parse(text, path)
    documents: list[Document] = []
    for index, record in enumerate(records):
        content, metadata = self._render(record)
        documents.append(
            Document(
                content=content,
                metadata={**base, **metadata, "record_index": index},
                source=path,
                mimetype="application/json",
            )
        )
    return documents

CSVLoader

CSVLoader(
    *,
    mode: str = "row",
    content_columns: list[str] | None = None,
    metadata_columns: list[str] | None = None,
    delimiter: str | None = None,
    max_rows: int | None = None,
    **config: Any
)

Bases: Loader

Loads delimited data files.

Parameters:

Name	Type	Description	Default
mode	str	"row" makes one document per row (right for records — support tickets, product listings); "table" makes one document for the whole file (right for small reference tables the model should see whole).	'row'
content_columns	list[str] | None	Columns included in the content. None uses all.	None
metadata_columns	list[str] | None	Columns copied into metadata.	None
delimiter	str | None	Field separator. Inferred from the extension when omitted.	None
max_rows	int | None	Stop after this many rows. None reads everything.	None
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Example

import tempfile, pathlib p = pathlib.Path(tempfile.mkdtemp()) / "d.csv" _ = p.write_text("name,role\nada,engineer\ngrace,admiral\n") docs = CSVLoader().load(p) len(docs), "ada" in docs[0].content (2, True)

Source code in src\windlass\providers\loaders\text.py

def __init__(
    self,
    *,
    mode: str = "row",
    content_columns: list[str] | None = None,
    metadata_columns: list[str] | None = None,
    delimiter: str | None = None,
    max_rows: int | None = None,
    **config: Any,
) -> None:
    if mode not in {"row", "table"}:
        raise ValueError("mode must be 'row' or 'table'")
    super().__init__(**config)
    self.mode = mode
    self.content_columns = content_columns
    self.metadata_columns = metadata_columns
    self.delimiter = delimiter
    self.max_rows = max_rows

aload_source async

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

Read a delimited file.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path or a bytes payload.	required

Returns:

Type	Description
list[Document]	One document per row, or one for the whole table.

Raises:

Type	Description
IngestionError	When the file cannot be parsed.

Source code in src\windlass\providers\loaders\text.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Read a delimited file.

    Args:
        source: A path or a bytes payload.

    Returns:
        One document per row, or one for the whole table.

    Raises:
        IngestionError: When the file cannot be parsed.
    """
    raw = self._read_bytes(source)
    text = _decode(raw, self.encoding)
    path = str(source) if not isinstance(source, bytes) else None
    base = self._base_metadata(source) if not isinstance(source, bytes) else {}
    delimiter = self.delimiter or _sniff_delimiter(path, text)

    try:
        reader = csv.DictReader(io.StringIO(text), delimiter=delimiter)
        rows = list(reader)[: self.max_rows] if self.max_rows else list(reader)
    except csv.Error as exc:
        raise IngestionError(f"Could not parse {path or 'input'} as CSV: {exc}") from exc

    if not rows:
        return []

    if self.mode == "table":
        # Render as a Markdown table: models read the header/column
        # relationship far more reliably from this than from raw CSV.
        headers = [h for h in rows[0] if h is not None]
        lines = [
            "| " + " | ".join(headers) + " |",
            "| " + " | ".join("---" for _ in headers) + " |",
        ]
        lines += [
            "| " + " | ".join(str(row.get(h, "") or "") for h in headers) + " |" for row in rows
        ]
        return [
            Document(
                content="\n".join(lines),
                metadata={**base, "rows": len(rows), "columns": headers},
                source=path,
                mimetype="text/csv",
            )
        ]

    documents: list[Document] = []
    for index, row in enumerate(rows):
        columns = self.content_columns or list(row.keys())
        content = "\n".join(f"{k}: {row.get(k, '')}" for k in columns if k)
        metadata = (
            {k: row[k] for k in self.metadata_columns if k in row}
            if self.metadata_columns
            else {}
        )
        documents.append(
            Document(
                content=content,
                metadata={**base, **metadata, "row_index": index},
                source=path,
                mimetype="text/csv",
            )
        )
    return documents