checkpoint.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/checkpoint.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_checkpoint_*.py"
"""
Operator CLI for registry JSON checkpoints.
 
Purpose
-------
Expose `python -m oc_chat.registry checkpoint export|import` for portable
registry state transfer.
 
Responsibilities
----------------
- Resolve registry paths through standard discovery.
- Wire export/import arguments to the checkpoint library.
- Format checkpoint results as human text or JSON.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - checkpoint command lifecycle events
    L3 - checkpoint result summaries
    L4 - argparse/dispatch tracing
 
Contracts
---------
- Import delegates signature verification and restore to `checkpoint.py`.
- Export does not mutate registry state.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.checkpoint (checkpoint implementation)
- oc_chat.registry.discovery (path resolution)
- oc_chat.registry (Registry facade passed to the checkpoint export)
- oc_chat.registry.backends.sqlite (SQLiteBackend opened for export and restored on import)
"""
from __future__ import annotations
 
import argparse
import json
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.checkpoint import (
    import_checkpoint,
    write_checkpoint,
)
from oc_chat.registry.discovery import discover_registry_location
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_checkpoint_subcommands(parser: argparse.ArgumentParser) -> None:
    """Attach ``checkpoint export|import`` subcommands."""
 
    parser.set_defaults(handler=_cmd_checkpoint_group, checkpoint_parser=parser)
    subcommands = parser.add_subparsers(dest="checkpoint_command")
 
    export = subcommands.add_parser("export", help="Export a JSON checkpoint.")
    export.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    export.add_argument("--out", required=True, help="Output checkpoint JSON path.")
    export.add_argument("--sign", action="store_true", help="Sign with HMAC-SHA256.")
    export.add_argument("--json", action="store_true")
    export.set_defaults(handler=_cmd_export)
 
    import_parser = subcommands.add_parser("import", help="Import a JSON checkpoint.")
    import_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    import_parser.add_argument(
        "--from",
        dest="from_file",
        required=True,
        help="Checkpoint JSON file to import.",
    )
    import_parser.add_argument(
        "--skip-signature-check",
        action="store_true",
        help="Allow importing an unsigned checkpoint.",
    )
    import_parser.add_argument("--json", action="store_true")
    import_parser.set_defaults(handler=_cmd_import)
 
 
def _cmd_checkpoint_group(args: argparse.Namespace) -> int:
    parser = args.checkpoint_parser
    parser.print_help()
    return 1
 
 
def _cmd_export(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    result = write_checkpoint(
        Registry(backend),
        out=Path(args.out),
        sign=args.sign,
    )
    payload = result.to_dict()
    if args.json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        print(f"checkpoint exported: {result.path}")
        print(f"sha256: {result.sha256}")
    return 0
 
 
def _cmd_import(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    result = import_checkpoint(
        backend,
        checkpoint_path=Path(args.from_file),
        skip_signature_check=args.skip_signature_check,
    )
    payload = result.to_dict()
    if args.json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        print(f"checkpoint imported: {result.path}")
        print(f"tables: {len(result.tables)}")
    return 0