39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
|
|
class StructuredOutputError(ValueError):
|
|
pass
|
|
|
|
|
|
def load_json_object(content: str, label: str) -> dict[str, Any]:
|
|
try:
|
|
data = json.loads(content)
|
|
except (json.JSONDecodeError, TypeError) as exc:
|
|
raise StructuredOutputError(f"{label}: invalid JSON: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise StructuredOutputError(f"{label}: expected JSON object")
|
|
return data
|
|
|
|
|
|
def load_json_schema(path: str | Path) -> dict[str, Any]:
|
|
return json.loads(Path(path).read_text())
|
|
|
|
|
|
def validate_json_object(
|
|
data: dict[str, Any],
|
|
schema: dict[str, Any],
|
|
label: str,
|
|
) -> dict[str, Any]:
|
|
errors = sorted(Draft202012Validator(schema).iter_errors(data), key=lambda error: error.path)
|
|
if errors:
|
|
first = errors[0]
|
|
location = ".".join(str(part) for part in first.absolute_path) or "<root>"
|
|
raise StructuredOutputError(f"{label}: schema violation at {location}: {first.message}")
|
|
return data
|