VaultKG (vaultkg.module)

vaultkg.module

module.py -- VaultKG, the KGModule for Obsidian-style Markdown vaults.

VaultKG(repo_root, db_path=None, vectors_path=None, *, exclude=None, **kwargs)

Bases: KGModule

Knowledge graph over one Markdown vault.

Artefacts live in <vault>/.vaultkg/; Obsidian ignores dot-folders, so they never show up as notes.

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Vault root.	required
exclude	list[str] | tuple[str, ...] | None	Directory names or glob patterns to skip when building, added to those in .vaultkg/config.yaml and to every dot-directory and node_modules.	None
kwargs	Any	Passed to :class:kg_utils.pipeline.KGModule (db_path, vectors_path, model...).	{}

Source code in src/vaultkg/module.py

def __init__(
    self,
    repo_root: str | Path,
    db_path: str | Path | None = None,
    vectors_path: str | Path | None = None,
    *,
    exclude: list[str] | tuple[str, ...] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(repo_root, db_path, vectors_path, **kwargs)
    configured = load_config(self.repo_root).get("exclude", [])
    # Arguments add to the stored list; order kept, duplicates dropped.
    self.exclude = tuple(dict.fromkeys([*configured, *(exclude or ())]))

analyze()

Markdown graph-health report. Never raises.

Source code in src/vaultkg/module.py

def analyze(self) -> str:
    """Markdown graph-health report. Never raises."""
    try:
        return render_report(self.health(), self.stats())
    except Exception as exc:  # noqa: BLE001 -- contract: never raise
        return f"# VaultKG Analysis\n\nAnalysis failed: {exc}\n"

health()

Graph-health figures (hubs, orphans, bridges, wanted pages...).

Source code in src/vaultkg/module.py

def health(self) -> VaultHealth:
    """Graph-health figures (hubs, orphans, bridges, wanted pages...)."""
    return vault_health(self.store.con)

links(node_id, *, direction='out', rel='', limit=50)

Links at a node: what it links to, or what links to it (backlinks).

A note's backlinks include links to any of its sections ([[Note#Heading]]), as Obsidian counts them.

Parameters:

Name	Type	Description	Default
node_id	str	Node id or vault path, as for :meth:node.	required
direction	str	"out" (the node is the source) or "in".	'out'
rel	str	One relation (LINKS_TO, SUPPORTS...), or "" for all.	''
limit	int	Links returned, 1-500.	50

Returns:

Type	Description
list[dict[str, Any]]	{rel, node, kind, name, evidence, via} per link, ordered by relation then node id. via is the node the link lands on: the node itself, or for a note's backlinks, the section a link points into.

Raises:

Type	Description
ValueError	On a bad id, direction or limit.

Source code in src/vaultkg/module.py

def links(
    self, node_id: str, *, direction: str = "out", rel: str = "", limit: int = 50
) -> list[dict[str, Any]]:
    """Links at a node: what it links to, or what links to it (backlinks).

    A note's backlinks include links to any of its sections
    (``[[Note#Heading]]``), as Obsidian counts them.

    :param node_id: Node id or vault path, as for :meth:`node`.
    :param direction: ``"out"`` (the node is the source) or ``"in"``.
    :param rel: One relation (``LINKS_TO``, ``SUPPORTS``...), or ``""`` for all.
    :param limit: Links returned, 1-500.
    :return: ``{rel, node, kind, name, evidence, via}`` per link, ordered
        by relation then node id. ``via`` is the node the link lands on:
        the node itself, or for a note's backlinks, the section a link
        points into.
    :raises ValueError: On a bad id, direction or limit.
    """
    nid = normalize_node_id(node_id)
    if direction not in ("out", "in"):
        raise ValueError(f"direction must be 'out' or 'in', got {direction!r}")
    bounded_int("limit", limit, 1, MAX_LIMIT)
    near, far = ("src", "dst") if direction == "out" else ("dst", "src")
    where = f"e.{near} = ?"
    params: list[Any] = [nid]
    if direction == "in" and nid.startswith("note:"):
        # A link to one of the note's sections is a backlink to the note,
        # as Obsidian and analyze() count it. Headings are matched by id
        # prefix as a range, which needs no escaping of % or _ in paths.
        # The note's own heading tree (CONTAINS among its sections) is
        # structure inside the note, not a link to it.
        lo = "heading:" + nid.removeprefix("note:") + "#"
        hi = lo[:-1] + chr(ord("#") + 1)
        where = (
            "(e.dst = ? OR (e.dst >= ? AND e.dst < ?)) "
            "AND e.src != ? AND NOT (e.src >= ? AND e.src < ?)"
        )
        params = [nid, lo, hi, nid, lo, hi]
    if rel:
        where += " AND e.rel = ?"
        params.append(rel.strip().upper())
    sql = (
        f"SELECT e.rel, e.{far}, n.kind, n.name, e.evidence, e.{near} FROM edges e "
        f"LEFT JOIN nodes n ON n.id = e.{far} WHERE {where} "
        f"ORDER BY e.rel, e.{far}, e.{near} LIMIT ?"
    )
    params.append(limit)
    return [
        {
            "rel": r[0],
            "node": r[1],
            "kind": r[2],
            "name": r[3],
            "evidence": json.loads(r[4]) if r[4] else {},
            "via": r[5],
        }
        for r in self.store.con.execute(sql, params)
    ]

node(node_id)

One node, by id or by the note's vault path.

Parameters:

Name	Type	Description	Default
node_id	str	note:wiki/X.md, heading:..., tag:..., or a bare vault path such as wiki/X.md or wiki/X.	required

Returns:

Type	Description
dict[str, Any] | None	The node, or None if there is none.

Raises:

Type	Description
ValueError	On an empty or over-long id.

Source code in src/vaultkg/module.py

def node(self, node_id: str) -> dict[str, Any] | None:
    """One node, by id or by the note's vault path.

    :param node_id: ``note:wiki/X.md``, ``heading:...``, ``tag:...``, or a
        bare vault path such as ``wiki/X.md`` or ``wiki/X``.
    :return: The node, or ``None`` if there is none.
    :raises ValueError: On an empty or over-long id.
    """
    return self.store.node(normalize_node_id(node_id))

pack(q, *, k=8, hop=1, rels=None, **kw)

Like :meth:query, with each note or section's source text attached.

Parameters:

Name	Type	Description	Default
q	str	Natural-language query.	required
k	int	Seed hits.	8
hop	int	Expansion hops.	1
rels	tuple[str, ...] | None	Relations to expand along; default every relation present.	None

Returns:

Type	Description
SnippetPack	The pack.

Source code in src/vaultkg/module.py

def pack(
    self, q: str, *, k: int = 8, hop: int = 1, rels: tuple[str, ...] | None = None, **kw: Any
) -> SnippetPack:
    """Like :meth:`query`, with each note or section's source text attached.

    :param q: Natural-language query.
    :param k: Seed hits.
    :param hop: Expansion hops.
    :param rels: Relations to expand along; default every relation present.
    :return: The pack.
    """
    return super().pack(q, k=k, hop=hop, rels=rels or self.relations(), **kw)

query(q, *, k=8, hop=1, rels=None, **kw)

Semantic seeding plus link expansion.

Parameters:

Name	Type	Description	Default
q	str	Natural-language query.	required
k	int	Seed hits.	8
hop	int	Expansion hops along links, embeds, tags and headings.	1
rels	tuple[str, ...] | None	Relations to expand along; default every relation present.	None

Returns:

Type	Description
QueryResult	Ranked nodes and the edges among them.

Source code in src/vaultkg/module.py

def query(
    self, q: str, *, k: int = 8, hop: int = 1, rels: tuple[str, ...] | None = None, **kw: Any
) -> QueryResult:
    """Semantic seeding plus link expansion.

    :param q: Natural-language query.
    :param k: Seed hits.
    :param hop: Expansion hops along links, embeds, tags and headings.
    :param rels: Relations to expand along; default every relation present.
    :return: Ranked nodes and the edges among them.
    """
    return super().query(q, k=k, hop=hop, rels=rels or self.relations(), **kw)

relations()

Every relation in the built graph, structural ones first.

Typed links are whatever keys a vault uses (SUPPORTS, UP...), so query expansion follows the relations actually present rather than a fixed list.

Source code in src/vaultkg/module.py

def relations(self) -> tuple[str, ...]:
    """Every relation in the built graph, structural ones first.

    Typed links are whatever keys a vault uses (``SUPPORTS``, ``UP``...),
    so query expansion follows the relations actually present rather than
    a fixed list.
    """
    present = {r[0] for r in self.store.con.execute("SELECT DISTINCT rel FROM edges")}
    return tuple(r for r in STRUCTURAL_RELS if r in present) + tuple(
        sorted(present - set(STRUCTURAL_RELS))
    )

stats()

Store counts, plus note/link figures a vault owner cares about.

Returns:

Type	Description
dict[str, Any]	total_nodes, total_edges, node_counts, edge_counts, meaningful_nodes (everything but unresolved targets), plus notes, headings, tags, unresolved.

Source code in src/vaultkg/module.py

def stats(self) -> dict[str, Any]:
    """Store counts, plus note/link figures a vault owner cares about.

    :return: ``total_nodes``, ``total_edges``, ``node_counts``,
        ``edge_counts``, ``meaningful_nodes`` (everything but unresolved
        targets), plus ``notes``, ``headings``, ``tags``, ``unresolved``.
    """
    s = {k: v for k, v in super().stats().items() if k not in _CODE_GRAPH_KEYS}
    nc = s.get("node_counts", {})
    s["notes"] = nc.get("note", 0)
    s["headings"] = nc.get("heading", 0)
    s["tags"] = nc.get("tag", 0)
    s["unresolved"] = nc.get("symbol", 0)
    return s

load_config(repo_root)

Read <vault>/.vaultkg/config.yaml.

The file is optional. exclude is a list of folder names or glob patterns, the same values vaultkg build --exclude takes. raw_dir is the vault-relative folder vaultkg add writes sources into (default :data:DEFAULT_RAW_DIR).

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Vault root.	required

Returns:

Type	Description
dict[str, Any]	{"exclude": [...]} plus "raw_dir" when set; empty when the file is absent.

Raises:

Type	Description
ValueError	The file is not a mapping, has an unknown key, exclude is not a list of non-empty strings, or raw_dir is not a relative folder inside the vault.

Source code in src/vaultkg/module.py

def load_config(repo_root: str | Path) -> dict[str, Any]:
    """Read ``<vault>/.vaultkg/config.yaml``.

    The file is optional. ``exclude`` is a list of folder names or glob
    patterns, the same values ``vaultkg build --exclude`` takes. ``raw_dir``
    is the vault-relative folder ``vaultkg add`` writes sources into
    (default :data:`DEFAULT_RAW_DIR`).

    :param repo_root: Vault root.
    :return: ``{"exclude": [...]}`` plus ``"raw_dir"`` when set; empty when
        the file is absent.
    :raises ValueError: The file is not a mapping, has an unknown key,
        ``exclude`` is not a list of non-empty strings, or ``raw_dir`` is not
        a relative folder inside the vault.
    """
    path = Path(repo_root) / CONFIG_FILE
    if not path.is_file():
        return {}
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    if not isinstance(data, dict):
        raise ValueError(f"{path}: expected a mapping, got {type(data).__name__}")
    unknown = sorted(set(data) - _CONFIG_KEYS)
    if unknown:
        raise ValueError(f"{path}: unknown keys {unknown}; allowed: {sorted(_CONFIG_KEYS)}")
    exclude = data.get("exclude") or []
    if not isinstance(exclude, list) or not all(isinstance(x, str) and x for x in exclude):
        raise ValueError(f"{path}: exclude must be a list of non-empty strings")
    out: dict[str, Any] = {"exclude": exclude}
    if "raw_dir" in data:
        raw_dir = data["raw_dir"]
        parts = Path(raw_dir).parts if isinstance(raw_dir, str) else ()
        if not parts or Path(raw_dir).is_absolute() or ".." in parts:
            raise ValueError(f"{path}: raw_dir must be a relative folder inside the vault")
        out["raw_dir"] = raw_dir
    return out

normalize_node_id(raw)

Normalise a node id supplied from outside (a CLI argument, an MCP call).

Agents echo ids back with backticks or quotes, and people type a note's path rather than its id, so wiki/X, wiki/X.md and note:wiki/X.md all name the same note.

Parameters:

Name	Type	Description	Default
raw	str	The id as given.	required

Returns:

Type	Description
str	A node id with one of the extractor's prefixes.

Raises:

Type	Description
ValueError	If empty after stripping, or longer than 500 characters.

Source code in src/vaultkg/module.py

def normalize_node_id(raw: str) -> str:
    """Normalise a node id supplied from outside (a CLI argument, an MCP call).

    Agents echo ids back with backticks or quotes, and people type a note's
    path rather than its id, so ``wiki/X``, ``wiki/X.md`` and ``note:wiki/X.md``
    all name the same note.

    :param raw: The id as given.
    :return: A node id with one of the extractor's prefixes.
    :raises ValueError: If empty after stripping, or longer than 500 characters.
    """
    nid = raw.strip().strip("`'\"").strip()
    if not nid:
        raise ValueError("node_id must not be empty")
    if len(nid) > MAX_ID_LEN:
        raise ValueError(f"node_id must be at most {MAX_ID_LEN} characters, got {len(nid)}")
    if nid.startswith(_ID_PREFIXES):
        return nid
    path = nid.lstrip("/")
    return "note:" + (path if path.lower().endswith(".md") else path + ".md")