Files
troupe-standup/wizard-work/spark/json_schema_validator.py
T

89 lines
2.6 KiB
Python

#!/usr/bin/env python3
"""
Quick JSON Schema Validator
Usage: python3 json_schema_validator.py <schema_file.json> <data_file.json>
Or: python3 json_schema_validator.py <schema_file.json> (reads JSON from stdin)
Exits with code 0 if valid, 1 if invalid, prints errors to stderr.
"""
import json
import sys
from typing import Any
try:
from jsonschema import validate, ValidationError, Draft202012Validator
HAS_JSONSCHEMA = True
except ImportError:
HAS_JSONSCHEMA = False
print("Installing jsonschema...", file=sys.stderr)
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install", "jsonschema", "-q"])
from jsonschema import validate, ValidationError, Draft202012Validator
HAS_JSONSCHEMA = True
def load_json(path: str = None) -> Any:
"""Load JSON from file or stdin."""
if path and path != "-":
with open(path, 'r') as f:
return json.load(f)
return json.load(sys.stdin)
def validate_json(data: Any, schema: Any, schema_path: str = "<stdin>") -> list[str]:
"""Validate data against schema. Returns list of errors (empty if valid)."""
if not HAS_JSONSCHEMA:
return ["jsonschema package not available"]
errors = []
# Collect all validation errors (not just first)
validator = Draft202012Validator(schema)
for error in validator.iter_errors(data):
path = " => ".join(str(p) for p in error.absolute_path) or "<root>"
errors.append(f"{path}: {error.message}")
return errors
def main():
if len(sys.argv) < 2:
print(__doc__, file=sys.stderr)
sys.exit(1)
schema_path = sys.argv[1]
data_path = sys.argv[2] if len(sys.argv) > 2 else None
try:
schema = load_json(schema_path if schema_path != "-" else None)
except FileNotFoundError:
print(f"Error: Schema file not found: {schema_path}", file=sys.stderr)
sys.exit(1)
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in schema: {e}", file=sys.stderr)
sys.exit(1)
try:
data = load_json(data_path)
except FileNotFoundError:
print(f"Error: Data file not found: {data_path}", file=sys.stderr)
sys.exit(1)
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in data: {e}", file=sys.stderr)
sys.exit(1)
errors = validate_json(data, schema, schema_path)
if errors:
print("Validation FAILED:", file=sys.stderr)
for err in errors:
print(f" - {err}", file=sys.stderr)
sys.exit(1)
else:
print("Validation PASSED")
sys.exit(0)
if __name__ == "__main__":
main()