Ingestion API

All ingesters accept a GraphStore instance and return an IngestionResult dataclass.

from navegador.graph import GraphStore
from navegador.ingest import RepoIngester, KnowledgeIngester, WikiIngester, PlanopticonIngester

IngestionResult

@dataclass
class IngestionResult:
    nodes_created: int
    nodes_updated: int
    edges_created: int
    files_processed: int
    errors: list[str]
    duration_seconds: float

RepoIngester

Parses a source tree and writes code layer nodes and edges.

class RepoIngester:
    def __init__(self, store: GraphStore) -> None: ...

    def ingest(
        self,
        path: str | Path,
        *,
        clear: bool = False,
        incremental: bool = False,
        redact: bool = False,
        monorepo: bool = False,
    ) -> IngestionResult: ...

    def ingest_file(
        self,
        path: str | Path,
        *,
        redact: bool = False,
    ) -> IngestionResult: ...

Usage

store = GraphStore.sqlite(".navegador/navegador.db")
ingester = RepoIngester(store)

# full repo ingest
result = ingester.ingest("./src")
print(f"{result.nodes_created} nodes, {result.edges_created} edges")

# incremental ingest — only reprocesses files whose content hash has changed
result = ingester.ingest("./src", incremental=True)

# incremental: single file
result = ingester.ingest_file("./src/auth/service.py")

# wipe + rebuild
result = ingester.ingest("./src", clear=True)

# redact sensitive content (strips tokens, passwords, keys from string literals)
result = ingester.ingest("./src", redact=True)

# monorepo — traverse workspace sub-packages
result = ingester.ingest("./monorepo", monorepo=True)

Supported languages

Adding a new language parser

1. Install the tree-sitter grammar: pip install tree-sitter-<lang>
2. Subclass navegador.ingest.base.LanguageParser:

from navegador.ingest.base import LanguageParser, ParseResult

class RubyParser(LanguageParser):
    language = "ruby"
    extensions = [".rb"]

    def parse(self, source: str, file_path: str) -> ParseResult:
        # use self.tree_sitter_language to build the tree
        # return ParseResult with nodes and edges
        ...

1. Register in navegador/ingest/registry.py:

from .ruby import RubyParser
PARSERS["ruby"] = RubyParser

RepoIngester dispatches to registered parsers by file extension.

Framework enrichers

After parsing, RepoIngester runs framework-specific enrichers that annotate nodes with framework context. Enrichers are discovered automatically based on what frameworks are detected in the repo.

KnowledgeIngester

Writes knowledge layer nodes. Wraps the navegador add commands programmatically.

class KnowledgeIngester:
    def __init__(self, store: GraphStore) -> None: ...

    def add_concept(
        self,
        name: str,
        *,
        description: str = "",
        domain: str = "",
        status: str = "",
    ) -> str: ...  # returns node ID

    def add_rule(
        self,
        name: str,
        *,
        description: str = "",
        domain: str = "",
        severity: str = "info",
        rationale: str = "",
    ) -> str: ...

    def add_decision(
        self,
        name: str,
        *,
        description: str = "",
        domain: str = "",
        rationale: str = "",
        alternatives: str = "",
        date: str = "",
        status: str = "proposed",
    ) -> str: ...

    def add_person(
        self,
        name: str,
        *,
        email: str = "",
        role: str = "",
        team: str = "",
    ) -> str: ...

    def add_domain(
        self,
        name: str,
        *,
        description: str = "",
    ) -> str: ...

    def annotate(
        self,
        code_name: str,
        *,
        node_type: str = "Function",
        concept: str = "",
        rule: str = "",
    ) -> None: ...

Usage

store = GraphStore.sqlite(".navegador/navegador.db")
ingester = KnowledgeIngester(store)

ingester.add_domain("Payments", description="Payment processing and billing")
ingester.add_concept("Idempotency", domain="Payments",
    description="Operations safe to retry without side effects")
ingester.add_rule("RequireIdempotencyKey",
    domain="Payments", severity="critical",
    rationale="Card networks retry on timeout")
ingester.annotate("process_payment", node_type="Function",
    concept="Idempotency", rule="RequireIdempotencyKey")

WikiIngester

Fetches GitHub wiki pages and writes WikiPage nodes.

class WikiIngester:
    def __init__(self, store: GraphStore) -> None: ...

    def ingest_repo(
        self,
        repo: str,
        *,
        token: str = "",
        use_api: bool = False,
    ) -> IngestionResult: ...

    def ingest_dir(
        self,
        path: str | Path,
    ) -> IngestionResult: ...

Usage

import os
store = GraphStore.sqlite(".navegador/navegador.db")
ingester = WikiIngester(store)

# from GitHub API
result = ingester.ingest_repo("myorg/myrepo", token=os.environ["GITHUB_TOKEN"])

# from local clone
result = ingester.ingest_dir("./myrepo.wiki")

PlanopticonIngester

Ingests Planopticon knowledge graph output into the knowledge layer.

class PlanopticonIngester:
    def __init__(self, store: GraphStore) -> None: ...

    def ingest(
        self,
        path: str | Path,
        *,
        input_type: str = "auto",
        source: str = "",
    ) -> IngestionResult: ...

    def ingest_manifest(
        self,
        path: str | Path,
        *,
        source: str = "",
    ) -> IngestionResult: ...

    def ingest_kg(
        self,
        path: str | Path,
        *,
        source: str = "",
    ) -> IngestionResult: ...

    def ingest_interchange(
        self,
        path: str | Path,
        *,
        source: str = "",
    ) -> IngestionResult: ...

    def ingest_batch(
        self,
        path: str | Path,
        *,
        source: str = "",
    ) -> IngestionResult: ...

input_type values: "auto", "manifest", "kg", "interchange", "batch".

See Planopticon guide for format details and entity mapping.

Export and import

Navegador can export the full graph (or a subset) to JSONL for backup, migration, or sharing. The JSONL format is one JSON object per line, where each object is either a node or an edge.

navegador export > graph.jsonl
navegador export --nodes-only > nodes.jsonl
navegador import graph.jsonl

Python API:

from navegador.graph import GraphStore

store = GraphStore.sqlite(".navegador/navegador.db")

# export
with open("graph.jsonl", "w") as f:
    store.export_jsonl(f)

# import into a new store
new_store = GraphStore.sqlite(".navegador/new.db")
with open("graph.jsonl") as f:
    new_store.import_jsonl(f)

Schema migrations

When upgrading navegador, run navegador migrate before re-ingesting to apply schema changes (new node properties, new edge types, index updates):

navegador migrate

Migrations are idempotent — safe to run multiple times. The migration state is stored in the graph itself under a _MigrationState node.

Python API:

from navegador.graph import GraphStore, migrate

store = GraphStore.sqlite(".navegador/navegador.db")
migrate(store)   # applies any pending migrations