db.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/db.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_operator_tools_and_guards.py"
"""
Database operator commands for the orchestration registry.
 
Purpose
-------
Expose read-only database health commands beneath
``python -m oc_chat.registry db``.
 
Responsibilities
----------------
- Wire the ``db integrity-check`` command.
- Reserve future database operation names with explicit v0.2.x stubs.
- Format structured integrity reports for humans or JSON consumers.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- ``integrity-check`` does not mutate logical registry state.
- Reserved commands exit 2 with a clear not-implemented message.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.discovery (registry path discovery)
"""
from __future__ import annotations
 
import argparse
import json
import sys
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
RESERVED_DB_COMMANDS = ("backup", "restore", "prune", "archive-log", "restore-log")
 
 
def add_db_subcommands(db_parser: argparse.ArgumentParser) -> None:
    """Attach the M11.1c db subcommands to ``db_parser``."""
 
    db_parser.set_defaults(handler=_cmd_db_group, db_parser=db_parser)
    subcommands = db_parser.add_subparsers(dest="db_command")
 
    integrity = subcommands.add_parser(
        "integrity-check",
        help="Run SQLite physical, FK, and append-only integrity checks.",
    )
    integrity.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    integrity.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured integrity report as JSON.",
    )
    integrity.set_defaults(handler=_cmd_integrity_check)
 
    for name in RESERVED_DB_COMMANDS:
        reserved = subcommands.add_parser(
            name,
            help="Reserved for a later registry phase.",
        )
        reserved.set_defaults(handler=_cmd_reserved_db)
 
 
def _cmd_db_group(args: argparse.Namespace) -> int:
    parser = args.db_parser
    parser.print_help()
    return 1
 
 
def _cmd_integrity_check(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path, check_capabilities=False)
    report = backend.integrity_check()
    if args.json:
        print(json.dumps(report, indent=2, sort_keys=True))
    else:
        print(format_integrity_report(report))
    return 0 if report.get("ok") is True else 1
 
 
def _cmd_reserved_db(args: argparse.Namespace) -> int:
    print(f"db {args.db_command} not implemented in v0.2.x", file=sys.stderr)
    return 2
 
 
def format_integrity_report(report: Mapping[str, object]) -> str:
    """Return a compact human-readable integrity report."""
 
    errors = report.get("errors", [])
    if report.get("ok") is True:
        return "integrity ok"
    if not isinstance(errors, list):
        return "integrity failed: malformed error report"
    lines = [f"integrity failed: {len(errors)} error(s)"]
    for error in errors:
        if isinstance(error, Mapping):
            kind = error.get("kind", "unknown")
            details = error.get("details", "")
            lines.append(f"  - {kind}: {details}")
        else:
            lines.append(f"  - {error}")
    return "\n".join(lines)