Harden registry API and schema validation

railiance_fabric/schema_validation.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import warnings
+from pathlib import Path
+from typing import Any
+
+import jsonschema
+
+try:
+    from referencing import Registry, Resource
+except ModuleNotFoundError:
+    Registry = None
+    Resource = None
+
+from .loader import load_yaml
+
+
+def schema_registry(schemas_dir: Path) -> Any:
+    if Registry is None or Resource is None:
+        return None
+
+    resources: list[tuple[str, Any]] = []
+    for path in sorted(schemas_dir.glob("*.schema.yaml")):
+        schema = load_yaml(path)
+        if not isinstance(schema, dict):
+            continue
+        resource = Resource.from_contents(schema)
+        resources.append((path.resolve().as_uri(), resource))
+        schema_id = schema.get("$id")
+        if isinstance(schema_id, str):
+            resources.append((schema_id, resource))
+    return Registry().with_resources(resources)
+
+
+def draft202012_validator(schema_path: Path) -> jsonschema.Draft202012Validator:
+    schema = load_yaml(schema_path)
+    if not isinstance(schema, dict):
+        raise ValueError(f"schema must be a mapping: {schema_path}")
+    registry = schema_registry(schema_path.parent)
+    if registry is None:
+        return _legacy_validator(schema_path, schema)
+    return jsonschema.Draft202012Validator(
+        schema,
+        registry=registry,
+    )
+
+
+def _legacy_validator(
+    schema_path: Path,
+    schema: dict[str, Any],
+) -> jsonschema.Draft202012Validator:
+    store: dict[str, Any] = {}
+    for path in sorted(schema_path.parent.glob("*.schema.yaml")):
+        loaded_schema = load_yaml(path)
+        store[path.resolve().as_uri()] = loaded_schema
+        if isinstance(loaded_schema, dict):
+            schema_id = loaded_schema.get("$id")
+            if isinstance(schema_id, str):
+                store[schema_id] = loaded_schema
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", category=DeprecationWarning)
+        resolver = jsonschema.RefResolver(
+            base_uri=schema_path.resolve().as_uri(),
+            referrer=schema,
+            store=store,
+        )
+        return jsonschema.Draft202012Validator(schema, resolver=resolver)