70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""CLI client for the governed local writer service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Send one governed-local-writer request")
|
|
parser.add_argument("--socket", required=True, help="AF_UNIX socket path")
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--request", help="Request JSON string")
|
|
group.add_argument("--request-file", help="Path to JSON request file")
|
|
return parser.parse_args()
|
|
|
|
|
|
def _load_request_text(args: argparse.Namespace) -> str:
|
|
if args.request is not None:
|
|
return args.request
|
|
request_path = os.fspath(args.request_file)
|
|
with open(request_path, "r", encoding="utf-8") as request_file:
|
|
return request_file.read()
|
|
|
|
|
|
def send_request(socket_path: str, request_text: str) -> dict[str, object]:
|
|
"""Open one Unix socket, send one NDJSON request, and parse one NDJSON response."""
|
|
payload = json.loads(request_text)
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("request must be a JSON object")
|
|
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
|
|
client.connect(socket_path)
|
|
client.sendall((json.dumps(payload) + "\n").encode("utf-8"))
|
|
response_chunks: list[str] = []
|
|
while True:
|
|
chunk = client.recv(1024 * 1024).decode("utf-8")
|
|
if not chunk:
|
|
break
|
|
response_chunks.append(chunk)
|
|
if "\n" in chunk:
|
|
break
|
|
response_text = "".join(response_chunks)
|
|
|
|
response_line = response_text.split("\n", 1)[0].strip()
|
|
if not response_line:
|
|
raise RuntimeError("empty response")
|
|
response = json.loads(response_line)
|
|
if not isinstance(response, dict):
|
|
raise RuntimeError("invalid response")
|
|
return response
|
|
|
|
|
|
def main() -> None:
|
|
args = _parse_args()
|
|
request_text = _load_request_text(args)
|
|
response = send_request(args.socket, request_text)
|
|
print(json.dumps(response, separators=(",", ":"), sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as exc: # pragma: no cover - CLI entrypoint behavior
|
|
print(json.dumps({"status": "error", "error": str(exc)}), file=sys.stderr)
|
|
raise SystemExit(1)
|