product-54354138-1/tools/example.py

"""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())