Draft 7, passing all but 4 (refRemote) tests.

Author: Julian

referencing/_core.py

@@ -1,14 +1,13 @@
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Iterable, Sequence
 from urllib.parse import unquote, urldefrag, urljoin
 
 from attrs import evolve, field
 from pyrsistent import m, plist, s
 from pyrsistent.typing import PList, PMap, PSet
 
 from referencing._attrs import define, frozen
-from referencing.jsonschema import DynamicAnchor, id_of
 from referencing.typing import Anchor as AnchorType, Schema, Specification
 
 
@@ -19,11 +18,10 @@ class UnidentifiedResource(Exception):
 @frozen
 class Anchor:
 
-    uri: str
     name: str
     resource: Schema
 
-    def resolve(self, dynamic_scope, uri) -> tuple[Schema, str]:
+    def resolve(self, resolver, uri) -> tuple[Schema, str]:
         return self.resource, uri
 
 
@@ -34,7 +32,15 @@ class OpaqueSpecification:
     In particular, they have no subresources.
     """
 
-    def subresources_of(self, resource: Schema):
+    def id_of(self, resource):
+        if resource is True or resource is False:
+            return
+        return resource.get("$id")  # REMOVEME
+
+    def anchors_in(self, resource):
+        return ()
+
+    def subresources_of(self, resource):
         return ()
 
 
@@ -45,7 +51,7 @@ class Registry:
         default=m(),
         repr=lambda value: f"({len(value)} entries)",
     )
-    _uncrawled: PSet[str] = s()
+    _uncrawled: PSet[str] = field(default=s(), repr=False)
     _specification: Specification = OpaqueSpecification()
 
     def update(self, *registries: Registry) -> Registry:
@@ -58,7 +64,7 @@ def update(self, *registries: Registry) -> Registry:
         )
 
     def with_resource(self, resource) -> Registry:
-        uri = id_of(resource)
+        uri = self._specification.id_of(resource)
         if uri is None:
             raise UnidentifiedResource(resource)
         return self.with_identified_resource(uri=uri, resource=resource)
@@ -77,17 +83,23 @@ def with_resources(self, pairs) -> Registry:
             ), (uri, self._contents[uri], resource)
             contents = contents.set(uri, (resource, m()))
 
-            id = id_of(resource)
+            id = self._specification.id_of(resource)
             if id is not None:
                 contents = contents.set(id, (resource, m()))
 
             uncrawled = uncrawled.add(uri)
         return evolve(self, contents=contents, uncrawled=uncrawled)
 
-    def with_anchor(self, anchor: AnchorType) -> Registry:
-        resource, anchors = self._contents[anchor.uri]
-        new = resource, anchors.set(anchor.name, anchor)
-        return evolve(self, contents=self._contents.set(anchor.uri, new))
+    def with_anchors(
+        self,
+        uri: str,
+        anchors: Iterable[AnchorType],
+    ) -> Registry:
+        assert "#" not in uri, uri
+        resource, old = self._contents[uri]
+        new = old.update({anchor.name: anchor for anchor in anchors})
+        contents = self._contents.set(uri, (resource, new))
+        return evolve(self, contents=contents)
 
     def resource_at(self, uri: str) -> tuple[Schema, Registry]:
         at_uri = self._contents.get(uri)
@@ -108,28 +120,16 @@ def _crawl(self) -> Registry:
             if resource is True or resource is False:
                 continue
 
-            uri = urljoin(base_uri, resource.get("$id", ""))
+            uri = urljoin(base_uri, self._specification.id_of(resource) or "")
             if uri != base_uri:
                 registry = registry.with_identified_resource(
                     uri=uri,
                     resource=resource,
                 )
 
-            anchor = resource.get("$anchor")
-            if anchor is not None:
-                registry = registry.with_anchor(
-                    Anchor(uri=uri, name=anchor, resource=resource),
-                )
+            anchors = self._specification.anchors_in(resource)
+            registry = registry.with_anchors(uri=uri, anchors=anchors)
 
-            dynamic_anchor = resource.get("$dynamicAnchor")
-            if dynamic_anchor is not None:
-                registry = registry.with_anchor(
-                    DynamicAnchor(
-                        uri=uri,
-                        name=dynamic_anchor,
-                        resource=resource,
-                    ),
-                )
             resources.extend(
                 (uri, each)
                 for each in self._specification.subresources_of(resource)
@@ -138,7 +138,7 @@ def _crawl(self) -> Registry:
         return evolve(registry, uncrawled=s())
 
     def resolver(self, root, specification) -> Resolver:
-        uri = id_of(root) or ""
+        uri = self._specification.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)
@@ -167,15 +167,12 @@ def lookup(self, ref: str) -> tuple[Schema, Resolver]:
                 target = target[segment]  # type: ignore # this can't be a bool
         elif fragment:
             anchor = registry.anchors_at(uri=uri)[fragment]
-            target, uri = anchor.resolve(
-                dynamic_scope=self.dynamic_scope(),
-                uri=uri,
-            )
+            target, uri = anchor.resolve(resolver=self, uri=uri)
 
         return target, self.evolve(base_uri=uri, registry=registry)
 
     def with_root(self, root) -> Resolver:
-        maybe_relative = id_of(root)
+        maybe_relative = self._registry._specification.id_of(root)
         if maybe_relative is None:
             return self
 

referencing/jsonschema.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 from referencing._attrs import frozen
+from referencing._core import Anchor
 from referencing.typing import Schema
 
 
@@ -14,6 +15,20 @@ class Draft202012:
     _SUBRESOURCE_ITEMS = {"allOf"}
     _SUBRESOURCE_VALUES = {"$defs", "properties"}
 
+    def id_of(self, resource):
+        if resource is True or resource is False:
+            return
+        return resource.get("$id")
+
+    def anchors_in(self, resource):
+        anchor = resource.get("$anchor")
+        if anchor is not None:
+            yield Anchor(name=anchor, resource=resource)
+
+        dynamic_anchor = resource.get("$dynamicAnchor")
+        if dynamic_anchor is not None:
+            yield DynamicAnchor(name=dynamic_anchor, resource=resource)
+
     def subresources_of(self, resource):
         for each in self._SUBRESOURCE:
             if each in resource:
@@ -29,19 +44,19 @@ def subresources_of(self, resource):
 @frozen
 class DynamicAnchor:
 
-    uri: str
     name: str
     resource: Schema
 
-    def resolve(self, dynamic_scope, uri):
+    def resolve(self, resolver, uri):
         last = self.resource
-        for resource, anchors in dynamic_scope:
+        for resource, anchors in resolver.dynamic_scope():
             anchor = anchors.get(self.name)
             if isinstance(anchor, DynamicAnchor):
                 last = anchor.resource
             elif "$ref" not in resource:
                 break
-        return last, id_of(last) or ""  # FIXME: consider when this can be None
+        id = resolver._registry._specification.id_of(last) or ""  # FIXME
+        return last, id
 
 
 class Draft201909:
@@ -50,6 +65,16 @@ class Draft201909:
     _SUBRESOURCE_ITEMS = {"allOf"}
     _SUBRESOURCE_VALUES = {"$defs", "properties"}
 
+    def id_of(self, resource):
+        if resource is True or resource is False:
+            return
+        return resource.get("$id")
+
+    def anchors_in(self, resource):
+        anchor = resource.get("$anchor")
+        if anchor is not None:
+            yield Anchor(name=anchor, resource=resource)
+
     def subresources_of(self, resource):
         for each in self._SUBRESOURCE:
             if each in resource:
@@ -70,7 +95,39 @@ def subresources_of(self, resource):
             yield items
 
 
-def id_of(resource) -> str | None:
-    if resource is True or resource is False:
-        return None
-    return resource.get("$id")
+class Draft7:
+
+    _SUBRESOURCE = {"not"}
+    _SUBRESOURCE_ITEMS = {"allOf"}
+    _SUBRESOURCE_VALUES = {"definitions", "properties"}
+
+    def id_of(self, resource):
+        if resource is True or resource is False or "$ref" in resource:
+            return None
+        id = resource.get("$id")
+        if id is not None and not id.startswith("#"):
+            return id
+
+    def anchors_in(self, resource):
+        anchor = resource.get("$id", "")
+        if anchor.startswith("#"):
+            yield Anchor(name=anchor[1:], resource=resource)
+
+    def subresources_of(self, resource):
+        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

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Iterable, Protocol, Union
+from typing import TYPE_CHECKING, Any, Iterable, Protocol, Union
 
 try:
     from collections.abc import Mapping
@@ -9,29 +9,34 @@
 except TypeError:
     from typing import Mapping
 
+if TYPE_CHECKING:
+    from referencing._core import Resolver
+
 
 ObjectSchema = Mapping[str, Any]
 Schema = Union[bool, ObjectSchema]
 
 
 class Anchor(Protocol):
-    @property
-    def uri(self) -> str:
-        ...
-
     @property
     def name(self) -> str:
         ...
 
-    def resolve(
-        self,
-        dynamic_scope: Iterable[tuple[Schema, Anchor]],
-        uri: str,
-    ) -> tuple[Schema, str]:
+    def resolve(self, resolver: Resolver, uri: str) -> tuple[Schema, str]:
         pass
 
 
 class Specification(Protocol):
+    def id_of(self, resource: Schema) -> str | None:
+        """
+        The URI ID of the given resource.
+        """
+
+    def anchors_in(self, resource: ObjectSchema) -> Iterable[Anchor]:
+        """
+        All (non-recursively nested) anchors inside the given resource.
+        """
+
     def subresources_of(self, resource: ObjectSchema) -> Iterable[Schema]:
         """
         All (non-recursively nested) resources inside the given resource.