Enough additional cowboy coding to get draft 2019-09 working too.

Author: Julian

referencing/__init__.py

@@ -1 +1,6 @@
-from referencing._core import Registry  # noqa: F401
+from referencing._core import (  # noqa: F401
+    OpaqueSpecification as _Opaque,
+    Registry,
+)
+
+OPAQUE_SPECIFICATION = _Opaque()

referencing/_core.py

@@ -1,17 +1,12 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Union
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
 from urllib.parse import unquote, urldefrag, urljoin
 
 from pyrsistent import m, plist, s
 from pyrsistent.typing import PList, PMap, PSet
 
-try:
-    Mapping[str, str]
-except TypeError:
-    from typing import Mapping
-
 
 class UnsupportedSubclassing(Exception):
     @classmethod
@@ -29,6 +24,8 @@ class UnidentifiedResource(Exception):
 
 if TYPE_CHECKING:
     from attrs import define, evolve, field, frozen
+
+    from referencing.typing import AnchorType, Schema, Specification
 else:
     from attrs import define as _define, evolve, field, frozen as _frozen
 
@@ -41,9 +38,6 @@ def frozen(cls):
         return _frozen(cls)
 
 
-Schema = Union[bool, Mapping[str, Any]]
-
-
 @frozen
 class Anchor:
 
@@ -73,7 +67,15 @@ def resolve(self, dynamic_scope, uri) -> tuple[Schema, str]:
         return last, id_of(last) or ""  # FIXME: consider when this can be None
 
 
-AnchorType = Union[Anchor, DynamicAnchor]
+class OpaqueSpecification:
+    """
+    A non-specification `Specification` which treats resources opaquely.
+
+    In particular, they have no subresources.
+    """
+
+    def subresources_of(self, resource: Schema):
+        return ()
 
 
 @frozen
@@ -83,7 +85,8 @@ class Registry:
         default=m(),
         repr=lambda value: f"({len(value)} entries)",
     )
-    _uncrawled: PSet[str] = field(default=s())
+    _uncrawled: PSet[str] = s()
+    _specification: Specification = OpaqueSpecification()
 
     def update(self, *registries: Registry) -> Registry:
         contents = (each._contents for each in registries)
@@ -167,27 +170,17 @@ def _crawl(self) -> Registry:
                         resource=resource,
                     ),
                 )
-
-            resources.extend(  # TODO: delay finding anchors in subresources...
-                (uri, resource[k]) for k in SUBRESOURCE if k in resource
-            )
-            resources.extend(
-                (uri, subresource)
-                for k in SUBRESOURCE_VALUES
-                if k in resource
-                for subresource in resource[k].values()
-            )
             resources.extend(
-                (uri, subresource)
-                for k in SUBRESOURCE_ITEMS
-                if k in resource
-                for subresource in resource[k]
+                (uri, each)
+                for each in self._specification.subresources_of(resource)
+                if each is not True and each is not False
             )
         return evolve(registry, uncrawled=s())
 
-    def resolver(self, root) -> Resolver:
+    def resolver(self, root, specification) -> Resolver:
         uri = id_of(root) or ""
         registry = self.with_identified_resource(uri=uri, resource=root)
+        registry = evolve(registry, specification=specification)
         return Resolver(base_uri=uri, registry=registry)
 
 
@@ -243,11 +236,6 @@ def dynamic_scope(self):
             yield resource, self._registry.anchors_at(uri)
 
 
-SUBRESOURCE = {"items", "not"}
-SUBRESOURCE_ITEMS = {"allOf"}
-SUBRESOURCE_VALUES = {"$defs", "properties"}
-
-
 def id_of(resource) -> str | None:
     if resource is True or resource is False:
         return None

referencing/jsonschema.py

@@ -0,0 +1,54 @@
+"""
+Referencing implementations for JSON Schema specs (historic & current).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from referencing.typing import ObjectSchema
+
+
+class Draft202012:
+
+    _SUBRESOURCE = {"items", "not"}
+    _SUBRESOURCE_ITEMS = {"allOf"}
+    _SUBRESOURCE_VALUES = {"$defs", "properties"}
+
+    def subresources_of(self, resource: ObjectSchema):
+        for each in self._SUBRESOURCE:
+            if each in resource:
+                yield resource[each]
+        for each in self._SUBRESOURCE_ITEMS:
+            if each in resource:
+                yield from resource[each]
+        for each in self._SUBRESOURCE_VALUES:
+            if each in resource:
+                yield from resource[each].values()
+
+
+class Draft201909:
+
+    _SUBRESOURCE = {"not"}
+    _SUBRESOURCE_ITEMS = {"allOf"}
+    _SUBRESOURCE_VALUES = {"$defs", "properties"}
+
+    def subresources_of(self, resource: ObjectSchema):
+        for each in self._SUBRESOURCE:
+            if each in resource:
+                yield resource[each]
+        for each in self._SUBRESOURCE_ITEMS:
+            if each in resource:
+                yield from resource[each]
+        for each in self._SUBRESOURCE_VALUES:
+            if each in resource:
+                yield from resource[each].values()
+
+        items = resource.get("items")
+        if items is None:
+            return
+        elif isinstance(items, list):
+            yield from items
+        else:
+            yield items

referencing/typing.py

@@ -0,0 +1,21 @@
+from typing import Any, Iterable, Protocol, Union
+
+try:
+    from collections.abc import Mapping
+
+    Mapping[str, str]
+except TypeError:
+    from typing import Mapping
+
+from referencing._core import Anchor, DynamicAnchor
+
+ObjectSchema = Mapping[str, Any]
+Schema = Union[bool, ObjectSchema]
+AnchorType = Union[Anchor, DynamicAnchor]
+
+
+class Specification(Protocol):
+    def subresources_of(self, resource: ObjectSchema) -> Iterable[Schema]:
+        """
+        All (non-recursively nested) resources inside the given resource.
+        """