zettelkasten.datasets

zettelkasten.datasets

Dataset resolution: the numbers behind a dataset note.

A dataset note keeps its numbers out of the markdown body. The data block in its frontmatter records where the bytes live and how to read them::

data:
  backend: sidecar | dvc | api    # optional; inferred when omitted
  format: csv | json | parquet    # optional; inferred from the path suffix
  content_hash: "sha256:..."      # stable identity + cache key (optional)
  schema:                         # optional column declaration
    - {name: date,  dtype: date}
    - {name: close, dtype: float}
  path: prices.csv                # sidecar/dvc: a file in the source folder
  fetch: {provider: url, params: {url: "https://..."}}   # api: fetch spec

Resolution is deliberately unified. A path covers both a small committed sidecar file and a large DVC-tracked file — the only difference is whether a <path>.dvc pointer sits next to it: if the data file itself is absent we run dvc pull to materialize it, then read. A fetch spec covers a live API pull whose bytes are cached under <ANGELO_DIR>/zettel/datasets/<hash>.<ext> (gitignored and re-derivable, exactly like the full-text cache in ingest.py).

This module never mutates a note; it only reads and normalizes. Writing dataset notes is the caller's job (server.add_note / the dataset MCP tool).

DatasetTable dataclass

A normalized in-memory table resolved from a dataset note.

Source code in zettelkasten/datasets.py

@dataclass
class DatasetTable:
    """A normalized in-memory table resolved from a dataset note."""

    columns: list[str]
    rows: list[list[Any]]
    num_rows: int
    backend: str
    format: str
    content_hash: str = ""
    schema: list[dict[str, Any]] = field(default_factory=list)
    source_path: str = ""
    warning: str = ""
    # The true sha256 of the resolved bytes (empty when nothing resolved). Unlike
    # ``content_hash`` (which echoes the note's DECLARED hash when present, for
    # display), this is always the hash of the bytes actually read — the value a
    # snapshot restamp pins into the note.
    actual_hash: str = ""

    def slice(self, offset: int = 0, limit: int | None = None) -> "DatasetTable":
        """Return a copy with ``rows`` windowed for pagination.

        ``num_rows`` is preserved as the FULL count so the caller can report the
        total independently of the returned page.
        """
        start = max(0, int(offset or 0))
        window = self.rows[start:] if limit is None else self.rows[start : start + int(limit)]
        return DatasetTable(
            columns=self.columns,
            rows=window,
            num_rows=self.num_rows,
            backend=self.backend,
            format=self.format,
            content_hash=self.content_hash,
            schema=self.schema,
            source_path=self.source_path,
            warning=self.warning,
            actual_hash=self.actual_hash,
        )

    def to_dict(self, *, include_rows: bool = True) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "columns": self.columns,
            "num_rows": self.num_rows,
            "backend": self.backend,
            "format": self.format,
            "content_hash": self.content_hash,
            "schema": self.schema,
            "source_path": self.source_path,
        }
        if self.warning:
            payload["warning"] = self.warning
        if include_rows:
            payload["rows"] = self.rows
        return payload

slice

slice(offset: int = 0, limit: int | None = None) -> 'DatasetTable'

Return a copy with rows windowed for pagination.

num_rows is preserved as the FULL count so the caller can report the total independently of the returned page.

Source code in zettelkasten/datasets.py

def slice(self, offset: int = 0, limit: int | None = None) -> "DatasetTable":
    """Return a copy with ``rows`` windowed for pagination.

    ``num_rows`` is preserved as the FULL count so the caller can report the
    total independently of the returned page.
    """
    start = max(0, int(offset or 0))
    window = self.rows[start:] if limit is None else self.rows[start : start + int(limit)]
    return DatasetTable(
        columns=self.columns,
        rows=window,
        num_rows=self.num_rows,
        backend=self.backend,
        format=self.format,
        content_hash=self.content_hash,
        schema=self.schema,
        source_path=self.source_path,
        warning=self.warning,
        actual_hash=self.actual_hash,
    )

compute_bytes_hash

compute_bytes_hash(data: bytes) -> str

sha256 of raw bytes — the dataset's stable identity (matches ingest.py).

Source code in zettelkasten/datasets.py

def compute_bytes_hash(data: bytes) -> str:
    """sha256 of raw bytes — the dataset's stable identity (matches ingest.py)."""
    return hashlib.sha256(data).hexdigest()

dataset_cache_path

dataset_cache_path(content_hash: str, fmt: str) -> Path

Path to the cached bytes for a content hash + format.

Source code in zettelkasten/datasets.py

def dataset_cache_path(content_hash: str, fmt: str) -> Path:
    """Path to the cached bytes for a content hash + format."""
    ext = fmt if fmt in _FORMATS else "bin"
    return _datasets_cache_dir() / f"{_normalize_hash(content_hash)}.{ext}"

register_fetcher

register_fetcher(provider: str, fn: Any) -> None

Register an api dataset fetcher: fn(params: dict) -> bytes.

Source code in zettelkasten/datasets.py

def register_fetcher(provider: str, fn: Any) -> None:
    """Register an ``api`` dataset fetcher: ``fn(params: dict) -> bytes``."""
    _FETCHERS[str(provider).strip().lower()] = fn

resolve_dataset

resolve_dataset(note: Any, graph_dir: Path, *, force_live: bool = False) -> DatasetTable

Resolve a dataset note to a normalized in-memory table.

note is a :class:zettelkasten.graph.Note (or anything exposing a data dict). graph_dir is the source-folder path the note lives in (ZettelGraph.path), used to locate sidecar/DVC files. Never raises for routine problems — a missing file or unknown provider resolves to an empty table with a populated warning so callers can surface it in the UI.

force_live bypasses the api fetch cache so the resolved bytes reflect the CURRENT source (used by the snapshot restamp). Path backends always read the current file, so the flag is a no-op for them.

Source code in zettelkasten/datasets.py

def resolve_dataset(note: Any, graph_dir: Path, *, force_live: bool = False) -> DatasetTable:
    """Resolve a ``dataset`` note to a normalized in-memory table.

    ``note`` is a :class:`zettelkasten.graph.Note` (or anything exposing a
    ``data`` dict). ``graph_dir`` is the source-folder path the note lives in
    (``ZettelGraph.path``), used to locate sidecar/DVC files. Never raises for
    routine problems — a missing file or unknown provider resolves to an empty
    table with a populated ``warning`` so callers can surface it in the UI.

    ``force_live`` bypasses the ``api`` fetch cache so the resolved bytes reflect
    the CURRENT source (used by the snapshot restamp). Path backends always read
    the current file, so the flag is a no-op for them.
    """
    data = getattr(note, "data", None)
    if not isinstance(data, dict):
        data = {}

    declared_schema = data.get("schema") if isinstance(data.get("schema"), list) else []
    declared_cols = _schema_columns(data)
    fmt = _infer_format(data)

    backend = str(data.get("backend") or "").strip().lower()
    if not backend:
        backend = "api" if isinstance(data.get("fetch"), dict) else "sidecar"

    if backend in _PATH_BACKENDS:
        raw, rel, warning = _read_path_backend(data, graph_dir)
        source_path = rel
    elif backend == "api":
        raw, warning = _read_api_backend(data, fmt, force_live=force_live)
        source_path = str((data.get("fetch") or {}).get("provider") or "api")
    else:
        raw, warning, source_path = None, f"unknown dataset backend {backend!r}", ""

    if raw is None:
        return DatasetTable(
            columns=declared_cols,
            rows=[],
            num_rows=0,
            backend=backend,
            format=fmt,
            content_hash=_normalize_hash(str(data.get("content_hash") or "")),
            schema=list(declared_schema),
            source_path=source_path,
            warning=warning,
            actual_hash="",
        )

    header_opt = data.get("header")
    header_opt = bool(header_opt) if isinstance(header_opt, bool) else None
    tab = str(data.get("path") or "").lower().endswith(".tsv")
    columns, rows, parse_warning = _parse_table(
        raw, fmt, declared_cols, header=header_opt, tab=tab,
    )

    # Verify the declared content hash when present; a mismatch is a soft warning
    # (the bytes may have legitimately changed) rather than a hard failure.
    declared_hash = _normalize_hash(str(data.get("content_hash") or ""))
    actual_hash = compute_bytes_hash(raw)
    hash_warning = ""
    if declared_hash and declared_hash != actual_hash:
        hash_warning = (
            f"content_hash mismatch: note declares {declared_hash[:12]}…, "
            f"resolved bytes hash to {actual_hash[:12]}…"
        )

    warning = "; ".join(w for w in (parse_warning, hash_warning) if w)
    return DatasetTable(
        columns=columns,
        rows=rows,
        num_rows=len(rows),
        backend=backend,
        format=fmt,
        content_hash=declared_hash or actual_hash,
        schema=list(declared_schema),
        source_path=source_path,
        warning=warning,
        actual_hash=actual_hash,
    )

read_dataset_values

read_dataset_values(note: Any, graph_dir: Path, *, offset: int = 0, limit: int | None = 200) -> dict[str, Any]

Resolve a dataset note and return a paginated values payload.

Shape: {columns, rows, num_rows, offset, limit, backend, format, content_hash, schema, warning}. num_rows is the full row count; rows is the requested window.

Source code in zettelkasten/datasets.py

def read_dataset_values(
    note: Any,
    graph_dir: Path,
    *,
    offset: int = 0,
    limit: int | None = 200,
) -> dict[str, Any]:
    """Resolve a dataset note and return a paginated values payload.

    Shape: ``{columns, rows, num_rows, offset, limit, backend, format,
    content_hash, schema, warning}``. ``num_rows`` is the full row count; ``rows``
    is the requested window.
    """
    table = resolve_dataset(note, graph_dir)
    page = table.slice(offset=offset, limit=limit)
    payload = page.to_dict(include_rows=True)
    payload["offset"] = max(0, int(offset or 0))
    payload["limit"] = limit
    return payload