scripts.purge_stray_kg_databases

Reference › scripts › purge_stray_kg_databases

scripts/purge_stray_kg_databases.py drops pre-cutover residue databases from an orchgraph cluster: any database outside the KEEP set is reported, and with --apply dropped. A human runs it after a graph migration (see kg migration to orchgraph). It connects through create_graph and resolve_uri from orchestrata_ultimate.core.graph_factory.

Options: --uri (default resolve_uri()) and --apply; without it the run is a read-only report. _database_names lists the databases in SHOW DATABASES order, _node_count counts the nodes of each stray database, and _drop drops it. The output lists each stray database with its node count, then the remaining databases after a real drop. If nothing is stray it names the databases the cluster holds and exits 0.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

#!/usr/bin/env python3
"""Drop pre-cutover residue databases from an orchgraph cluster.

Since the 2026-07-26 cutover both graphs live on ONE orchgraph cluster and
isolation is the ``database=`` argument. Two things left residue databases
behind on the PRODUCTION cluster:

* the suite used to run its per-worker databases (``orchkg_w<N>``) there,
  before ``TEST_MEMGRAPH_URI`` pointed it at the dedicated test node;
* ``core.graph_factory.resolve_kg_database`` used to fall back to the literal
  ``"neo4j"``, so any caller with an unset ``KG_DATABASE`` opened a database
  of that name and wrote into it instead of failing. That fallback now
  raises, but the database it created remains.

Read-only by default: prints what it would drop, with node counts, and exits
0. ``--apply`` performs the drops. ``KEEP`` is the allowlist — everything
else on the cluster is residue by definition, so a database added later must
be added there rather than being silently spared.

Take a backup first (``scripts/backup_orchgraph.sh``); this is not
recoverable from the cluster itself.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import LiteralString, cast

from dotenv import load_dotenv
from neo4j import Driver

_repo_root = Path(__file__).resolve().parent.parent
_repo_src = _repo_root / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

# RUL-019: load .env at the entrypoint. This is load-bearing here, not
# boilerplate — a stale exported ``NEO4J_URI`` (a pre-cutover
# ``bolt://10.53.1.11:7687`` left in an interactive shell) otherwise wins,
# and this script would enumerate databases on the RETIRED Neo4j host
# instead of the orchgraph cluster. Loading .env sets ``KG_ROUTING_URI``,
# which ``resolve_uri`` prefers over ``NEO4J_URI``.
load_dotenv(_repo_root / ".env")

from orchestrata_ultimate.core.graph_factory import (  # noqa: E402
    NO_AUTH,
    create_graph,
    resolve_uri,
)

#: Databases that are NOT residue. ``memgraph`` is the shared semantic/RAG
#: graph (knowledge-wiki + orchestrata's semantic_graph both write it) and is
#: also the cluster's own default; ``orchkg`` is the operational KG.
KEEP: frozenset[str] = frozenset({"memgraph", "orchkg"})


def _database_names(driver: Driver) -> list[str]:
    """Every database on the cluster, in ``SHOW DATABASES`` order.

    The column name differs between engine versions, so the single returned
    value is taken positionally rather than by key.
    """
    with driver.session() as session:
        return [
            str(next(iter(record.values()))) for record in session.run("SHOW DATABASES")
        ]


def _node_count(driver: Driver, database: str) -> int:
    with driver.session(database=database) as session:
        record = session.run("MATCH (n) RETURN count(n) AS c").single()
    return int(record["c"]) if record else 0


def _drop(driver: Driver, database: str) -> None:
    with driver.session() as session:
        # DDL cannot parameterize database names, and ``database`` comes from
        # the cluster's own SHOW DATABASES listing (not user input) — the
        # cast is the driver-sanctioned escape for validated dynamic DDL.
        session.run(cast(LiteralString, f"DROP DATABASE {database}")).consume()


def main(argv: list[str] | None = None) -> int:
    """Report (or, with ``--apply``, drop) every database outside :data:`KEEP`."""
    parser = argparse.ArgumentParser(
        prog="purge_stray_kg_databases",
        description="Drop pre-cutover residue databases from an orchgraph cluster.",
    )
    parser.add_argument(
        "--uri", default=None, help="Bolt/routing URI (default: resolve_uri())."
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Perform the drops. Without it this is a read-only report.",
    )
    args = parser.parse_args(argv)

    driver = create_graph(uri=args.uri or resolve_uri(), auth=NO_AUTH)
    try:
        stray = [name for name in _database_names(driver) if name not in KEEP]
        if not stray:
            print(
                "No residue databases — cluster holds only: " + ", ".join(sorted(KEEP))
            )
            return 0

        verb = "Dropping" if args.apply else "Would drop"
        print(f"{verb} {len(stray)} residue database(s):")
        for name in stray:
            count = _node_count(driver, name)
            print(f"  - {name} ({count} node(s))")
            if args.apply:
                _drop(driver, name)

        if args.apply:
            remaining = sorted(_database_names(driver))
            print(f"Remaining databases: {', '.join(remaining)}")
        else:
            print("\nRead-only report — pass --apply to drop them.")
        return 0
    finally:
        driver.close()


if __name__ == "__main__":
    sys.exit(main())