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)