beam: add streaming JSONL schema validator with tests

wizard-work/beam/jsonl_validate.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""jsonl_validate.py — streaming JSONL schema validator.
+
+Usage:
+    jsonl_validate.py SCHEMA.json INPUT.jsonl [--max-errors N]
+
+SCHEMA is a small JSON document:
+    {
+      "required": ["id", "name"],
+      "fields": {
+        "id": "int",
+        "name": "str",
+        "tags": "list",
+        "meta": "object",
+        "score": "float",
+        "active": "bool",
+        "email": "str?"
+      }
+    }
+
+Type names: str, int, float, bool, list, object, any. Append '?' for optional
+(missing allowed). A field present with null is treated as missing for optional
+fields and as an error for required fields.
+
+Exit codes:
+    0  all records valid
+    1  one or more validation errors
+    2  usage / IO / schema error
+"""
+import argparse
+import json
+import sys
+from typing import Any, Dict, List, Tuple
+
+
+TYPE_CHECKS = {
+    "str": lambda v: isinstance(v, str),
+    "int": lambda v: isinstance(v, int) and not isinstance(v, bool),
+    "float": lambda v: (isinstance(v, (int, float)) and not isinstance(v, bool)),
+    "bool": lambda v: isinstance(v, bool),
+    "list": lambda v: isinstance(v, list),
+    "object": lambda v: isinstance(v, dict),
+    "any": lambda v: True,
+}
+
+
+def load_json(path: str) -> Any:
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def validate_schema(schema: Any) -> Tuple[bool, str]:
+    if not isinstance(schema, dict):
+        return False, "schema must be an object"
+    if "required" in schema and not (isinstance(schema["required"], list)
+                                    and all(isinstance(x, str) for x in schema["required"])):
+        return False, "'required' must be a list of strings"
+    if "fields" in schema and not isinstance(schema["fields"], dict):
+        return False, "'fields' must be an object"
+    fields = schema.get("fields", {})
+    for name, typ in fields.items():
+        if not isinstance(name, str):
+            return False, "field names must be strings"
+        base = typ[:-1] if isinstance(typ, str) and typ.endswith("?") else typ
+        if base not in TYPE_CHECKS:
+            return False, f"unknown type '{typ}' for field '{name}'"
+    return True, ""
+
+
+def check_value(typ: str, value: Any) -> bool:
+    optional = typ.endswith("?")
+    base = typ[:-1] if optional else typ
+    if value is None and optional:
+        return True
+    return TYPE_CHECKS[base](value)
+
+
+def validate_record(rec: Any, schema: dict) -> List[str]:
+    errs: List[str] = []
+    if not isinstance(rec, dict):
+        return ["record is not an object"]
+    required = set(schema.get("required", []))
+    fields = schema.get("fields", {})
+    for r in required:
+        if r not in rec or rec.get(r) is None:
+            errs.append(f"missing required field '{r}'")
+    for name, typ in fields.items():
+        if name in rec and rec[name] is not None:
+            if not check_value(typ, rec[name]):
+                errs.append(f"field '{name}': expected {typ}, got {type(rec[name]).__name__}")
+        elif name in rec and rec[name] is None:
+            optional = typ.endswith("?")
+            if name in required or not optional:
+                errs.append(f"field '{name}': null not allowed")
+    # unknown fields? warn-level: not an error, but collect as info? keep strict-ish: allow extras
+    return errs
+
+
+def main(argv: List[str]) -> int:
+    ap = argparse.ArgumentParser(description="Validate a JSONL file against a schema.")
+    ap.add_argument("schema")
+    ap.add_argument("input")
+    ap.add_argument("--max-errors", type=int, default=50)
+    args = ap.parse_args(argv)
+
+    try:
+        schema = load_json(args.schema)
+    except (OSError, json.JSONDecodeError) as e:
+        print(f"schema error: {e}", file=sys.stderr)
+        return 2
+    ok, msg = validate_schema(schema)
+    if not ok:
+        print(f"schema error: {msg}", file=sys.stderr)
+        return 2
+
+    total = 0
+    bad = 0
+    err_count = 0
+    stopped = False
+    try:
+        with open(args.input, "r", encoding="utf-8") as f:
+            for lineno, line in enumerate(f, 1):
+                line = line.strip()
+                if not line:
+                    continue
+                total += 1
+                try:
+                    rec = json.loads(line)
+                except json.JSONDecodeError as e:
+                    bad += 1
+                    err_count += 1
+                    print(f"line {lineno}: invalid JSON: {e}")
+                    if err_count >= args.max_errors:
+                        stopped = True
+                        break
+                    continue
+                errs = validate_record(rec, schema)
+                if errs:
+                    bad += 1
+                    for e in errs:
+                        err_count += 1
+                        print(f"line {lineno}: {e}")
+                        if err_count >= args.max_errors:
+                            stopped = True
+                            break
+                if stopped:
+                    break
+    except OSError as e:
+        print(f"input error: {e}", file=sys.stderr)
+        return 2
+
+    print(f"checked {total} records, {bad} invalid" + (" (truncated)" if stopped else ""))
+    return 1 if bad else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))