scripts.kg_schema

Reference › scripts › kg_schema

scripts/kg_schema.py is the command-line tool for the canonical knowledge-graph schema (SPEC-106 FR-001): it applies the canonical uniqueness constraints and checks that they are present. A human runs it when a graph is provisioned or the schema changes. It uses orchestrata_ultimate.knowledge_graph.canonical_schema, orchestrata_ultimate.knowledge_graph.architecture_writer, orchestrata_ultimate.core.graph and orchestrata_ultimate.core.graph_factory.

Global options are --uri (default from KG_ROUTING_URI or KG_URI), --user and --password; _driver_from_args opens the driver. The apply subcommand idempotently creates the canonical constraints (_cmd_apply); --dry-run prints the DDL without connecting, and --dialect selects neo4j or memgraph (default resolved from KG_ENGINE), the same identity table in each engine's grammar. The check subcommand (_cmd_check) verifies the constraints. main prints an exception as ERROR: and exits 1.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/kg_schema.py

#!/usr/bin/env python3
"""CLI for the canonical Knowledge-Graph schema (SPEC-106 FR-001).

Subcommands
-----------
apply   Idempotently create the canonical constraints on a live graph.
check   Conformance gate: report drift between the live graph and the canon
        (missing constraints, extra/unknown constraints, and labels whose
        live nodes lack their canonical key). Exits non-zero on any drift —
        wire this into CI/pre-commit once the migration lands (not this
        script's job to do that wiring).

Connection defaults to the ``KG_ROUTING_URI``/``KG_URI`` environment
variables. CLI flags override env vars, so the
test-fleet instances (e.g. the ``.13``/``.14`` per-worker Neo4j clones) can be
stamped explicitly via ``--uri`` without touching the shared default instance.
``apply --dry-run`` prints the DDL that would run without connecting to any
graph at all — use it to preview before stamping a real instance.

Examples
--------
    python scripts/kg_schema.py check
    python scripts/kg_schema.py apply --dry-run
    python scripts/kg_schema.py apply --uri bolt://10.53.1.13:7687
    python scripts/kg_schema.py check --uri bolt://10.53.1.14:7687
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Mapping
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors run_ast_pipeline.py;
# also lets static analysis resolve the import without the editable-install
# .pth machinery).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from neo4j import Driver, GraphDatabase  # noqa: E402

from orchestrata_ultimate.core.graph import BoltSyncRepository  # noqa: E402
from orchestrata_ultimate.core.graph_factory import (  # noqa: E402
    resolve_kg_dialect,
    resolve_uri,
)
from orchestrata_ultimate.knowledge_graph.architecture_writer import (  # noqa: E402
    check_adr_integrity,
)
from orchestrata_ultimate.knowledge_graph.canonical_schema import (  # noqa: E402
    apply,
    check,
    get_all_ddl,
)


def _driver_from_args(
    args: argparse.Namespace,
    env: Mapping[str, str] | None = None,
) -> Driver:
    env = env if env is not None else os.environ
    # resolve_uri() raises and names the missing key rather than building a
    # driver on an empty address. The old bolt://localhost:7687 default could
    # only turn a config gap into a connection-refused that reads like an
    # outage — nothing in this infrastructure listens on localhost.
    uri = resolve_uri(args.uri, env=env)
    # No credentials: every remaining graph endpoint negotiates no auth.
    return GraphDatabase.driver(uri, auth=None)


def _cmd_apply(args: argparse.Namespace, *, statements: list[str] | None = None) -> int:
    """Run ``apply`` for the CLI.

    ``statements`` defaults to the real canonical DDL (:func:`get_all_ddl`) —
    this is what every CLI invocation of ``kg_schema.py apply`` runs. The
    keyword-only override exists purely so tests can exercise this function's
    driver-connect/apply/close wiring end-to-end against a test-only DDL
    scope without ever touching a real canonical label on shared infra.
    """
    stmts = (
        statements
        if statements is not None
        # resolve_kg_dialect, not a literal "neo4j": an unset --dialect
        # must read KG_ENGINE at CALL time, or this CLI emits Neo4j DDL
        # against whatever engine is actually configured.
        else get_all_ddl(dialect=resolve_kg_dialect(getattr(args, "dialect", None)))
    )
    if not args.dry_run:
        driver = _driver_from_args(args)
        try:
            driver.verify_connectivity()
            apply(BoltSyncRepository(driver=driver), stmts)
        finally:
            driver.close()
    verb = "would apply" if args.dry_run else "Applied"
    print(f"{verb} {len(stmts)} canonical DDL statement(s):")
    for stmt in stmts:
        print(f"  - {stmt}")
    return 0


def _cmd_check(args: argparse.Namespace) -> int:
    driver = _driver_from_args(args)
    try:
        driver.verify_connectivity()
        repo = BoltSyncRepository(driver=driver)
        report = check(repo)
        for line in report.to_lines():
            print(line)
        # Ruling 1 (ADR immutability) gate half: flag any live Accepted
        # ArchitectureDecision whose body no longer matches its stamped
        # body_sha (plans/architecture-ontology-draft.md §9, wave
        # ontology-phase1) -- folded into the same conformance exit code.
        adr_findings = check_adr_integrity(repo)
        if adr_findings:
            print("ADR immutability violations (Accepted body_sha drift):")
            for adr_id in adr_findings:
                print(f"  - {adr_id}")
        return 0 if (report.ok and not adr_findings) else 1
    finally:
        driver.close()


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="kg_schema",
        description="Apply/check the canonical knowledge-graph schema (SPEC-106 FR-001).",
    )
    parser.add_argument(
        "--uri",
        default=None,
        help="Graph Bolt URI (default: KG_ROUTING_URI/KG_URI env var).",
    )
    parser.add_argument(
        "--user",
        default=None,
        help="Graph username (default: empty — endpoint negotiates no auth).",
    )
    parser.add_argument(
        "--password",
        default=None,
        help="Graph password (default: empty — endpoint negotiates no auth).",
    )

    sub = parser.add_subparsers(dest="command", required=True)
    apply_p = sub.add_parser(
        "apply", help="Idempotently create the canonical constraints."
    )
    apply_p.add_argument(
        "--dry-run",
        action="store_true",
        help="Print the DDL that would run without connecting to the graph.",
    )
    apply_p.add_argument(
        "--dialect",
        choices=["neo4j", "memgraph"],
        default=None,
        help="DDL dialect (default: resolved from KG_ENGINE at call time): "
        "neo4j or memgraph (orchgraph-kg cluster, "
        "migration plan Step 1.2 — same canonical identity table, that "
        "engine's grammar).",
    )
    sub.add_parser("check", help="Report drift between the live graph and the canon.")

    return parser


def main(argv: list[str] | None = None) -> None:
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "apply": _cmd_apply,
        "check": _cmd_check,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()