Automatically create metaschemas when accessing JSONSchema.metaschema

CHANGELOG.rst

@@ -11,6 +11,9 @@ Features:
 * JSON ``null``, ``true``, ``false`` literals
 * Relative JSON Pointer ``+``/``-`` array index adjustments
 * Unknown keywords are collected as annotations
+* Automatically create metaschemas as referenced by ``"$schema"``
+* Automatically detect the core vocabulary in metaschemas,
+  but allow specifying a default to use when none is detectable
 
 Experimental:
 
@@ -23,6 +26,11 @@ Breaking changes:
 * ``Catalog.add_format_validators()`` superseded by ``@format_validator`` / ``Catalog.enable_formats()``
 * Rename ``Catalog.session()`` context manager to ``Catalog.cache()``
 * Rename ``session`` parameter to ``cacheid`` in many places
+* Added ``Catalog.get_metaschema()``, analogous to ``Catalog.get_schema()``
+* ``Catalog.create_metashema()`` and ``Catalog.create_vocabulary()`` return the created instance
+* Rename ``core_vocabulary`` and ``core_vocabulary_uri`` parameters for
+  ``Metaschema.__init__()`` and ``Catalog.create_metaschema()`` respectively to
+  ``default_core_vocabulary`` and ``default_core_vocabulary_uri``
 * Rename public functions in the ``jsonpatch`` module
 * Rename ``*Applicator*`` keyword class mixins to ``*Subschema*``
 

jschon/catalog/__init__.py

@@ -137,16 +137,19 @@ def load_json(self, uri: URI) -> JSONCompatible:
 
         raise CatalogError(f'A source is not available for "{uri}"')
 
-    def create_vocabulary(self, uri: URI, *kwclasses: KeywordClass) -> None:
+    def create_vocabulary(self, uri: URI, *kwclasses: KeywordClass) -> Vocabulary:
         """Create a :class:`~jschon.vocabulary.Vocabulary` object, which
         may be used by a :class:`~jschon.vocabulary.Metaschema` to provide
         keyword classes used in schema construction.
 
         :param uri: the URI identifying the vocabulary
         :param kwclasses: the :class:`~jschon.vocabulary.Keyword` classes
             constituting the vocabulary
+
+        :returns: the newly created :class:`Vocabulary` instance
         """
         self._vocabularies[uri] = Vocabulary(uri, *kwclasses)
+        return self._vocabularies[uri]
 
     def get_vocabulary(self, uri: URI) -> Vocabulary:
         """Get a :class:`~jschon.vocabulary.Vocabulary` by its `uri`.
@@ -162,37 +165,72 @@ def get_vocabulary(self, uri: URI) -> Vocabulary:
     def create_metaschema(
             self,
             uri: URI,
-            core_vocabulary_uri: URI,
+            default_core_vocabulary_uri: Optional[URI] = None,
             *default_vocabulary_uris: URI,
             **kwargs: Any,
-    ) -> None:
+    ) -> Metaschema:
         """Create, cache and validate a :class:`~jschon.vocabulary.Metaschema`.
 
         :param uri: the URI identifying the metaschema
-        :param core_vocabulary_uri: the URI identifying the metaschema's
-            core :class:`~jschon.vocabulary.Vocabulary`
+        :param default_core_vocabulary_uri: the URI identifying the metaschema's
+            core :class:`~jschon.vocabulary.Vocabulary`, used in the absence
+            of a ``"$vocabulary"`` keyword in the metaschema JSON file, or
+            if a known core vocabulary is not present under ``"$vocabulary"``
         :param default_vocabulary_uris: default :class:`~jschon.vocabulary.Vocabulary`
             URIs, used in the absence of a ``"$vocabulary"`` keyword in the
             metaschema JSON file
         :param kwargs: additional keyword arguments to pass through to the
             :class:`~jschon.jsonschema.JSONSchema` constructor
+
+        :returns: the newly created :class:`Metaschema` instance
+
+        :raise CatalogError: if the metaschema is not valid
         """
         metaschema_doc = self.load_json(uri)
-        core_vocabulary = self.get_vocabulary(core_vocabulary_uri)
+        default_core_vocabulary = (
+            self.get_vocabulary(default_core_vocabulary_uri)
+            if default_core_vocabulary_uri
+            else None
+        )
         default_vocabularies = [
             self.get_vocabulary(vocab_uri)
             for vocab_uri in default_vocabulary_uris
         ]
         metaschema = Metaschema(
             self,
             metaschema_doc,
-            core_vocabulary,
+            default_core_vocabulary,
             *default_vocabularies,
             **kwargs,
             uri=uri,
         )
         if not metaschema.validate().valid:
-            raise CatalogError("The metaschema is invalid against itself")
+            raise CatalogError(
+                "The metaschema is invalid against its own metaschema "
+                f'"{metaschema_doc["$schema"]}"'
+            )
+        return metaschema
+
+    def get_metaschema(self, uri: URI) -> Metaschema:
+        """Get a metaschema identified by `uri` from a cache, or
+        load it from configured sources if not already cached.
+
+        Note that metaschemas that do not declare a known core vocabulary
+        in ``$vocabulary`` must first be created using :meth:`create_schema`.
+
+        :param uri: the URI identifying the metaschema
+
+        :raise CatalogError: if the object referenced by `uri` is not
+            a :class:`~jschon.vocabulary.Metaschema`, or if it is not valid
+        :raise JSONSchemaError: if the metaschema is loaded from sources
+            but no known core vocabulary is present in ``$vocabulary``
+        """
+        metaschema = self._schema_cache['__meta__'].get(uri)
+        if not metaschema:
+            metaschema = self.create_metaschema(uri)
+        if not isinstance(metaschema, Metaschema):
+            raise CatalogError(f"The schema referenced by {uri} is not a metaschema")
+        return metaschema
 
     def enable_formats(self, *format_attr: str) -> None:
         """Enable validation of the specified format attributes.

jschon/jsonschema.py

@@ -218,21 +218,15 @@ def parentschema(self) -> Optional[JSONSchema]:
                 return parent
             parent = parent.parent
 
-    @property
+    @cached_property
     def metaschema(self) -> Metaschema:
         """The schema's :class:`~jschon.vocabulary.Metaschema`."""
         from jschon.vocabulary import Metaschema
 
         if (uri := self.metaschema_uri) is None:
             raise JSONSchemaError("The schema's metaschema URI has not been set")
 
-        if not isinstance(
-                metaschema := self.catalog.get_schema(uri, cacheid='__meta__'),
-                Metaschema,
-        ):
-            raise JSONSchemaError(f"The schema referenced by {uri} is not a metachema")
-
-        return metaschema
+        return self.catalog.get_metaschema(uri)
 
     @property
     def metaschema_uri(self) -> Optional[URI]:

jschon/vocabulary/__init__.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import re
 import inspect
 from typing import Any, Dict, Mapping, Optional, Sequence, TYPE_CHECKING, Tuple, Type
 
 from jschon.json import JSON, JSONCompatible
 from jschon.jsonschema import JSONSchema, Result
+from jschon.exceptions import JSONSchemaError
 from jschon.uri import URI
 
 if TYPE_CHECKING:
@@ -30,17 +32,54 @@ class Metaschema(JSONSchema):
     :class:`Metaschema` is itself a subclass of :class:`~jschon.jsonschema.JSONSchema`,
     and may be used to validate any referencing schema.
     """
+    _CORE_VOCAB_RE = r'https://json-schema\.org/draft/[^/]*/vocab/core$'
 
     def __init__(
             self,
             catalog: Catalog,
             value: Mapping[str, JSONCompatible],
-            core_vocabulary: Vocabulary,
+            default_core_vocabulary: Optional[Vocabulary] = None,
             *default_vocabularies: Vocabulary,
             **kwargs: Any,
     ):
-        self.core_vocabulary: Vocabulary = core_vocabulary
+        """Initialize a :class:`Metaschema` instance from the given
+        schema-compatible `value`.
+
+        :param catalog: catalog instance or catalog name
+        :param value: a schema-compatible Python object
+        :param default_core_vocabulary: the the metaschema's
+            core :class:`~jschon.vocabulary.Vocabulary`, used in the absence
+            of a ``"$vocabulary"`` keyword in the metaschema JSON file, or
+            if a known core vocabulary is not present under ``"$vocabulary"``
+        :param default_vocabulary: default :class:`~jschon.vocabulary.Vocabulary`
+            instances, used in the absence of a ``"$vocabulary"`` keyword in the
+            metaschema JSON file
+        :param kwargs: additional keyword arguments to pass through to the
+            :class:`~jschon.jsonschema.JSONSchema` constructor
+
+        :raise JSONSchemaError: if no core vocabulary can be determined
+        :raise CatalogError: if the created metaschema is not valid
+        """
         self.default_vocabularies: Tuple[Vocabulary, ...] = default_vocabularies
+        self.core_vocabulary: Vocabulary = default_core_vocabulary
+
+        if vocabularies := value.get("$vocabulary"):
+            possible_cores = list(filter(
+                lambda v: re.match(self._CORE_VOCAB_RE, v),
+                vocabularies,
+            ))
+            if len(possible_cores) == 1:
+                self.core_vocabulary = catalog.get_vocabulary(URI(possible_cores[0]))
+            else:
+                raise JSONSchemaError(
+                    'Cannot determine unique known core vocabulary from '
+                    f'candidates "{vocabularies.keys()}"'
+                )
+        if self.core_vocabulary is None:
+            raise JSONSchemaError(
+                f'No core vocabulary in "$vocabulary": {value}, and no default provided'
+            )
+
         self.kwclasses: Dict[str, KeywordClass] = {}
         super().__init__(value, catalog=catalog, cacheid='__meta__', **kwargs)
 

tests/__init__.py

@@ -4,6 +4,10 @@
 metaschema_uri_2020_12 = URI("https://json-schema.org/draft/2020-12/schema")
 metaschema_uri_next = URI("https://json-schema.org/draft/next/schema")
 
+core_vocab_uri_2019_09 = URI("https://json-schema.org/draft/2019-09/vocab/core")
+core_vocab_uri_2020_12 = URI("https://json-schema.org/draft/2020-12/vocab/core")
+core_vocab_uri_next = URI("https://json-schema.org/draft/next/vocab/core")
+
 example_schema = {
     "$schema": "https://json-schema.org/draft/2020-12/schema",
     "$id": "dynamicRef8_main.json",

tests/data/meta_invalid.json

@@ -0,0 +1,8 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "https://example.com/meta_invalid",
+    "$vocabulary": {
+        "https://json-schema.org/draft/2020-12/vocab/core": true
+    },
+    "type": {"lol": "cats"}
+}

tests/data/meta_no_vocabs.json

@@ -0,0 +1,4 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "https://example.com/meta_no_vocabs"
+}

tests/data/meta_with_core.json

@@ -0,0 +1,7 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "https://example.com/meta_with_core",
+    "$vocabulary": {
+        "https://json-schema.org/draft/2020-12/vocab/core": true
+    }
+}

tests/test_catalog.py

@@ -2,6 +2,7 @@
 import pathlib
 import tempfile
 import uuid
+import itertools
 
 import pytest
 
@@ -16,11 +17,28 @@
     LocalSource,
     RemoteSource,
 )
-from tests import example_schema, metaschema_uri_2020_12
+from jschon.vocabulary import Metaschema, Keyword
+from tests import example_schema, metaschema_uri_2020_12, core_vocab_uri_2020_12
 
 json_example = {"foo": "bar"}
 
 
+@pytest.fixture
+def local_catalog():
+    catalog = create_catalog(
+        '2019-09', '2020-12', 'next',
+        name='local'
+    )
+    catalog.add_uri_source(
+        URI('https://example.com/'),
+        LocalSource(
+            pathlib.Path(__file__).parent / 'data',
+            suffix='.json',
+        ),
+    )
+    return catalog
+
+
 @pytest.fixture
 def new_catalog() -> Catalog:
     return Catalog(name=str(uuid.uuid4()))
@@ -126,6 +144,17 @@ def test_get_vocabulary(uri, is_known, catalog):
             catalog.get_vocabulary(URI(uri))
 
 
+def test_create_vocabulary(catalog):
+    class CustomKeyword(Keyword):
+        key = 'custom'
+
+    custom_uri = URI('https://example.com/custom')
+    custom_vocab = catalog.create_vocabulary(custom_uri, CustomKeyword)
+    assert custom_vocab.uri is custom_uri
+    assert custom_vocab.kwclasses == {CustomKeyword.key: CustomKeyword}
+    assert catalog.get_vocabulary(custom_uri) is custom_vocab
+
+
 @pytest.fixture
 def example_schema_uri():
     schema = JSONSchema(example_schema)
@@ -183,3 +212,67 @@ def test_metaschema_isolation():
     assert okay_schema.evaluate(JSON(True)).valid is True
     okay_schema = cached_schema(uri, {"$ref": str(metaschema_uri_2020_12)}, None)
     assert okay_schema.evaluate(JSON(True)).valid is True
+
+
+def test_get_metaschema_detect_core(local_catalog):
+    uri = URI('https://example.com/meta_with_core')
+    core_vocab = local_catalog.get_vocabulary(core_vocab_uri_2020_12)
+
+    m = local_catalog.get_metaschema(uri)
+    assert isinstance(m, Metaschema)
+    assert m['$id'].data == str(uri)
+    assert m.core_vocabulary.uri == core_vocab.uri
+    assert m.kwclasses == core_vocab.kwclasses
+
+    s = local_catalog.get_schema(uri)
+    assert isinstance(s, JSONSchema)
+    assert s is not m
+    assert s == m
+
+
+def test_get_metaschema_wrong_type(local_catalog):
+    uri = URI('https://example.com/meta_with_core')
+    non_meta = local_catalog.get_schema(uri)
+    local_catalog._schema_cache['__meta__'][uri] = non_meta
+    with pytest.raises(CatalogError, match='not a metaschema'):
+        local_catalog.get_metaschema(uri)
+
+
+def test_get_metaschema_invalid(local_catalog):
+    uri = URI('https://example.com/meta_invalid')
+    with pytest.raises(CatalogError, match='metaschema is invalid'):
+        local_catalog.create_metaschema(uri)
+
+
+def test_create_metaschema_no_vocabs(local_catalog):
+    class ExtraKeyword(Keyword):
+        key='extra'
+
+    uri = URI('https://example.com/meta_no_vocabs')
+    core_vocab = local_catalog.get_vocabulary(core_vocab_uri_2020_12)
+    applicator_vocab = local_catalog.get_vocabulary(
+        URI('https://json-schema.org/draft/2020-12/vocab/applicator')
+    )
+
+    extra_vocab = local_catalog.create_vocabulary(
+        URI('https://example.com/vocab/whatever'),
+        ExtraKeyword,
+    )
+
+    m = local_catalog.create_metaschema(
+        uri,
+        core_vocab.uri,
+        applicator_vocab.uri,
+        extra_vocab.uri,
+    )
+    assert isinstance(m, Metaschema)
+    assert m['$id'].data == str(uri)
+    assert m.core_vocabulary is core_vocab
+    assert m.kwclasses.keys() == frozenset(
+        itertools.chain.from_iterable([
+            v.kwclasses.keys() for v in
+            [core_vocab, applicator_vocab, extra_vocab]
+        ])
+    )
+    m1 = local_catalog.get_metaschema(uri)
+    assert m1 is m