46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
"""Validate NDJSON events against the application-owned wire schemas."""
|
|
|
|
import json
|
|
import sys
|
|
from importlib import import_module
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
from pydantic import BaseModel, TypeAdapter
|
|
|
|
BACKEND_ROOT = Path(__file__).parents[1] / "backend"
|
|
EVENT_CLASS_NAMES = {
|
|
"metadata": "MetadataEvent",
|
|
"track": "TrackEvent",
|
|
"warning": "WarningEvent",
|
|
"error": "ErrorEvent",
|
|
"done": "DoneEvent",
|
|
}
|
|
|
|
|
|
def validate_event(line: str) -> BaseModel:
|
|
"""Parse one complete line with exact fields and strict value types."""
|
|
schemas = _load_schemas()
|
|
payload = json.loads(line)
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("NDJSON event must be an object")
|
|
event_type = payload.get("type")
|
|
if not isinstance(event_type, str) or event_type not in EVENT_CLASS_NAMES:
|
|
raise ValueError("NDJSON event has an unknown type")
|
|
event_class = getattr(schemas, EVENT_CLASS_NAMES[event_type])
|
|
if set(payload) != set(event_class.model_fields):
|
|
raise ValueError(f"{event_type} event fields do not match the wire schema")
|
|
if event_type == "track":
|
|
track = payload.get("track")
|
|
track_class = schemas.TrackCard
|
|
if not isinstance(track, dict) or set(track) != set(track_class.model_fields):
|
|
raise ValueError("track card fields do not match the wire schema")
|
|
adapter: TypeAdapter[BaseModel] = TypeAdapter(schemas.StreamEvent)
|
|
return adapter.validate_json(line, strict=True)
|
|
|
|
|
|
def _load_schemas() -> ModuleType:
|
|
backend_path = str(BACKEND_ROOT)
|
|
if backend_path not in sys.path:
|
|
sys.path.insert(0, backend_path)
|
|
return import_module("app.api.schemas")
|