"""Placeholder toolkit entry point that loads and validates a JSON request. This increment only establishes the stable command-line surface: read a single JSON request document, validate its shape, and print one structured JSON envelope. Later modules extend the same document format and envelope without changing the invocation contract, so the module stays deterministic and works fully offline. Usage:: python -m tools.example examples/input.json """ from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any, Mapping, Sequence __all__ = [ "EXIT_INVALID_REQUEST", "EXIT_OK", "EXIT_USAGE", "RequestError", "build_envelope", "load_request", "main", "validate_request", ] EXIT_OK = 0 EXIT_INVALID_REQUEST = 1 EXIT_USAGE = 2 #: Fields every request document must provide. REQUIRED_FIELDS = ("version", "mode", "dry_run", "command", "arguments") #: Modes understood by this increment of the toolkit. SUPPORTED_MODES = frozenset({"validate"}) PROG = "tools.example" class RequestError(ValueError): """Raised when a request document is missing or malformed.""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code self.message = message def load_request(path: str | Path) -> Any: """Read and decode the JSON document stored at ``path``.""" try: raw = Path(path).read_text(encoding="utf-8") except FileNotFoundError as exc: raise RequestError("request_not_found", f"request document not found: {path}") from exc except IsADirectoryError as exc: raise RequestError("request_not_a_file", f"request document is not a file: {path}") from exc except OSError as exc: raise RequestError("request_unreadable", f"request document could not be read: {path}") from exc except UnicodeDecodeError as exc: raise RequestError("request_not_utf8", f"request document is not valid UTF-8: {path}") from exc try: return json.loads(raw) except json.JSONDecodeError as exc: raise RequestError( "request_not_json", f"request document is not valid JSON: {path} (line {exc.lineno}, column {exc.colno})", ) from exc def _require(mapping: Mapping[str, Any], key: str, expected: type | tuple[type, ...], label: str) -> Any: if key not in mapping: raise RequestError("request_field_missing", f"request field {label} is required") value = mapping[key] if isinstance(value, bool) and bool not in ( expected if isinstance(expected, tuple) else (expected,) ): raise RequestError("request_field_invalid", f"request field {label} must be {_describe(expected)}") if not isinstance(value, expected): raise RequestError("request_field_invalid", f"request field {label} must be {_describe(expected)}") return value def _describe(expected: type | tuple[type, ...]) -> str: names = expected if isinstance(expected, tuple) else (expected,) return " or ".join(sorted({name.__name__ for name in names})) def validate_request(document: Any) -> dict[str, Any]: """Validate ``document`` and return the normalised request mapping.""" if not isinstance(document, dict): raise RequestError("request_not_object", "request document must be a JSON object") for key in REQUIRED_FIELDS: if key not in document: raise RequestError("request_field_missing", f"request field '{key}' is required") version = _require(document, "version", int, "version") if version < 1: raise RequestError("request_field_invalid", "request field 'version' must be a positive integer") mode = _require(document, "mode", str, "mode") if mode not in SUPPORTED_MODES: supported = ", ".join(sorted(SUPPORTED_MODES)) raise RequestError( "request_mode_unsupported", f"request field 'mode' must be one of: {supported}", ) dry_run = _require(document, "dry_run", bool, "dry_run") if not dry_run: raise RequestError( "request_not_dry_run", "this increment only supports validate-only dry_run requests", ) command = _require(document, "command", str, "command") if not command.strip(): raise RequestError("request_field_invalid", "request field 'command' must not be empty") arguments = _require(document, "arguments", dict, "arguments") request_path = _require(arguments, "request", str, "arguments.request") if not request_path.strip(): raise RequestError("request_field_invalid", "request field 'arguments.request' must not be empty") return { "version": version, "mode": mode, "dry_run": dry_run, "command": command, "arguments": dict(arguments), } def build_envelope(request: Mapping[str, Any], source: str) -> dict[str, Any]: """Build the deterministic success envelope for a validated request.""" return { "status": "ok", "command": request["command"], "mode": request["mode"], "dry_run": request["dry_run"], "version": request["version"], "request": source, } def build_error_envelope(error: RequestError, source: str) -> dict[str, Any]: """Build the deterministic failure envelope for an invalid request.""" return { "status": "error", "error": {"code": error.code, "message": error.message}, "request": source, } def emit(envelope: Mapping[str, Any]) -> None: """Print exactly one JSON document to standard output.""" print(json.dumps(envelope, sort_keys=True, ensure_ascii=True)) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog=PROG, description=( "Load and validate a JSON request document. The command is offline " "and prints one structured JSON envelope." ), ) parser.add_argument( "request", metavar="REQUEST", help="path to the JSON request document (for example examples/input.json)", ) return parser def main(argv: Sequence[str] | None = None) -> int: """Run the placeholder CLI and return the process exit code.""" parser = _build_parser() args = parser.parse_args(list(sys.argv[1:] if argv is None else argv)) try: document = load_request(args.request) request = validate_request(document) except RequestError as error: emit(build_error_envelope(error, args.request)) return EXIT_INVALID_REQUEST emit(build_envelope(request, args.request)) return EXIT_OK if __name__ == "__main__": raise SystemExit(main())