25 lines
803 B
Python
25 lines
803 B
Python
import json
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import jsonschema
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
@lru_cache
|
|
def get_event_validator() -> Draft202012Validator:
|
|
path = Path(get_settings().event_schema_path)
|
|
if not path.is_file():
|
|
path = Path(__file__).resolve().parents[3] / "schemas" / "event-schema-v1.json"
|
|
schema = json.loads(path.read_text(encoding="utf-8"))
|
|
Draft202012Validator.check_schema(schema)
|
|
return Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER)
|
|
|
|
|
|
def validate_event_payload(payload: dict) -> list[str]:
|
|
validator = get_event_validator()
|
|
errors = sorted(validator.iter_errors(payload), key=lambda e: e.path)
|
|
return [e.message for e in errors]
|