Simple anchor support.

Author: Julian

docs/conf.py

@@ -45,6 +45,7 @@
     D=("data", "D"),
     ObjectSchema=("data", "ObjectSchema"),
     Schema=("data", "Schema"),
+    URI=("attr", "URI"),  # ?!?!?! Sphinx...
 )
 
 

referencing/__init__.py

@@ -1,6 +1,6 @@
 """
 Cross-specification, implementation-agnostic JSON referencing.
 """
-from referencing._core import Registry, Resource, Specification
+from referencing._core import Anchor, Registry, Resource, Specification
 
-__all__ = ["Registry", "Resource", "Specification"]
+__all__ = ["Anchor", "Registry", "Resource", "Specification"]

referencing/_core.py

@@ -10,7 +10,7 @@
 
 from referencing import exceptions
 from referencing._attrs import frozen
-from referencing.typing import URI, D, Mapping
+from referencing.typing import URI, Anchor as AnchorType, D, Mapping
 
 
 @frozen
@@ -32,13 +32,25 @@ class Specification(Generic[D]):
     #: the subresources themselves).
     subresources_of: Callable[[D], Iterable[D]]
 
+    #: Retrieve the anchors contained in the given document.
+    _anchors_in: Callable[
+        [Specification[D], D],
+        Iterable[AnchorType[D]],
+    ] = field(alias="anchors_in")
+
     #: An opaque specification where resources have no subresources
     #: nor internal identifiers.
     OPAQUE: ClassVar[Specification[Any]]
 
     def __repr__(self):
         return f"<Specification name={self.name!r}>"
 
+    def anchors_in(self, contents: D):
+        """
+        Retrieve the anchors contained in the given document.
+        """
+        return self._anchors_in(self, contents)
+
     def create_resource(self, contents: D) -> Resource[D]:
         """
         Create a resource which is interpreted using this specification.
@@ -50,6 +62,7 @@ def create_resource(self, contents: D) -> Resource[D]:
     name="opaque",
     id_of=lambda contents: None,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 
 
@@ -126,6 +139,12 @@ def subresources(self) -> Iterable[Resource[D]]:
             for each in self._specification.subresources_of(self.contents)
         )
 
+    def anchors(self) -> Iterable[AnchorType[D]]:
+        """
+        Retrieve this resource's (specification-specific) identifier.
+        """
+        return self._specification.anchors_in(self.contents)
+
     def pointer(self, pointer: str, resolver: Resolver[D]) -> Resolved[D]:
         """
         Resolve the given JSON pointer.
@@ -168,6 +187,7 @@ class Registry(Mapping[URI, Resource[D]]):
     """
 
     _resources: PMap[URI, Resource[D]] = field(default=m(), converter=pmap)  # type: ignore[reportUnknownArgumentType]  # noqa: E501
+    _anchors: PMap[tuple[URI, str], AnchorType[D]] = field(default=m())  # type: ignore[reportUnknownArgumentType]  # noqa: E501
     _uncrawled: PSet[URI] = field(default=s())  # type: ignore[reportUnknownArgumentType]  # noqa: E501
 
     def __getitem__(self, uri: URI) -> Resource[D]:
@@ -201,6 +221,12 @@ def __repr__(self) -> str:
             summary = f"{pluralized}"
         return f"<Registry ({size} {summary})>"
 
+    def anchor(self, uri: URI, name: str):
+        """
+        Retrieve the given anchor, which must already have been found.
+        """
+        return self._anchors[uri, name]
+
     def contents(self, uri: URI) -> D:
         """
         Retrieve the contents identified by the given URI.
@@ -212,17 +238,24 @@ def crawl(self) -> Registry[D]:
         Immediately crawl all added resources, discovering subresources.
         """
         resources = self._resources.evolver()
+        anchors = self._anchors.evolver()
         uncrawled = [(uri, resources[uri]) for uri in self._uncrawled]
         while uncrawled:
             uri, resource = uncrawled.pop()
+
             id = resource.id()
-            if id is None:
-                pass
-            else:
+            if id is not None:
                 uri = urljoin(uri, id)
                 resources[uri] = resource
+            for each in resource.anchors():
+                anchors.set((uri, each.name), each)
             uncrawled.extend((uri, each) for each in resource.subresources())
-        return evolve(self, resources=resources.persistent(), uncrawled=s())
+        return evolve(
+            self,
+            resources=resources.persistent(),
+            anchors=anchors.persistent(),
+            uncrawled=s(),
+        )
 
     def with_resource(self, uri: URI, resource: Resource[D]):
         """
@@ -315,7 +348,10 @@ def lookup(self, ref: URI) -> Resolved[D]:
 
                 if the reference isn't resolvable
         """
-        uri, fragment = urldefrag(urljoin(self._base_uri, ref))
+        if ref.startswith("#"):
+            uri, fragment = self._base_uri, ref[1:]
+        else:
+            uri, fragment = urldefrag(urljoin(self._base_uri, ref))
         registry = self._registry
         resource = registry.get(uri)
         if resource is None:
@@ -325,8 +361,37 @@ def lookup(self, ref: URI) -> Resolved[D]:
             except KeyError:
                 raise exceptions.Unresolvable(ref=ref) from None
 
-        resolver = evolve(self, registry=registry, base_uri=uri)
         if fragment.startswith("/"):
-            return resource.pointer(pointer=fragment, resolver=resolver)
+            return resource.pointer(
+                pointer=fragment,
+                resolver=evolve(self, registry=registry, base_uri=uri),
+            )
+
+        if fragment:
+            try:
+                anchor = registry.anchor(uri, fragment)
+            except LookupError:
+                registry = registry.crawl()
+                anchor = registry.anchor(uri, fragment)
+
+            resource = anchor.resolve()
+        return Resolved(
+            contents=resource.contents,
+            resolver=evolve(self, registry=registry, base_uri=uri),
+        )
+
 
-        return Resolved(contents=resource.contents, resolver=resolver)
+@frozen
+class Anchor(Generic[D]):
+    """
+    A simple anchor in a `Resource`.
+    """
+
+    name: str
+    resource: Resource[D]
+
+    def resolve(self):
+        """
+        Return the resource for this anchor.
+        """
+        return self.resource

referencing/jsonschema.py

@@ -7,7 +7,7 @@
 from collections.abc import Set
 from typing import Any, Iterable, Union
 
-from referencing import Registry, Resource, Specification
+from referencing import Anchor, Registry, Resource, Specification
 from referencing._attrs import frozen
 from referencing.typing import URI, Mapping
 
@@ -38,13 +38,30 @@ def _dollar_id(contents: Schema) -> URI | None:
 def _dollar_id_pre2019(contents: Schema) -> URI | None:
     if isinstance(contents, bool) or "$ref" in contents:
         return
-    return contents.get("$id")
+    id = contents.get("$id")
+    if id is not None and not id.startswith("#"):
+        return id
 
 
 def _legacy_id(contents: ObjectSchema) -> URI | None:
     return contents.get("id")
 
 
+def _legacy_anchor_in_id(
+    specification: Specification[ObjectSchema],
+    contents: ObjectSchema,
+) -> Iterable[Anchor[ObjectSchema]]:
+    id = contents.get("$id", "")
+    if not id.startswith("#"):
+        return []
+    return [
+        Anchor(
+            name=id[1:],
+            resource=specification.create_resource(contents),
+        ),
+    ]
+
+
 def _subresources_of(
     in_value: Set[str] = frozenset(),
     in_subvalues: Set[str] = frozenset(),
@@ -75,11 +92,13 @@ def subresources_of(resource: ObjectSchema) -> Iterable[ObjectSchema]:
     name="draft2020-12",
     id_of=_dollar_id,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 DRAFT201909 = Specification(
     name="draft2019-09",
     id_of=_dollar_id,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 DRAFT7 = Specification(
     name="draft-07",
@@ -89,21 +108,25 @@ def subresources_of(resource: ObjectSchema) -> Iterable[ObjectSchema]:
         in_subarray={"allOf", "anyOf", "oneOf"},
         in_subvalues={"definitions"},
     ),
+    anchors_in=_legacy_anchor_in_id,
 )
 DRAFT6 = Specification(
     name="draft-06",
     id_of=_dollar_id,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 DRAFT4 = Specification(
     name="draft-04",
     id_of=_legacy_id,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 DRAFT3 = Specification(
     name="draft-03",
     id_of=_legacy_id,
     subresources_of=lambda contents: [],
+    anchors_in=lambda specification, contents: [],
 )
 
 

referencing/tests/test_core.py

@@ -1,12 +1,19 @@
 import pytest
 
-from referencing import Registry, Resource, Specification, exceptions
+from referencing import Anchor, Registry, Resource, Specification, exceptions
 from referencing.jsonschema import DRAFT202012
 
 ID_AND_CHILDREN = Specification(
     name="id-and-children",
     id_of=lambda contents: contents.get("ID"),
     subresources_of=lambda contents: contents.get("children", []),
+    anchors_in=lambda specification, contents: [
+        Anchor(
+            name=name,
+            resource=specification.create_resource(contents=each),
+        )
+        for name, each in contents.get("anchors", {}).items()
+    ],
 )
 
 
@@ -91,6 +98,30 @@ def test_crawl_finds_a_subresource(self):
         expected = ID_AND_CHILDREN.create_resource({"ID": child_id, "foo": 12})
         assert registry.crawl()[child_id] == expected
 
+    def test_crawl_finds_anchors_with_id(self):
+        resource = ID_AND_CHILDREN.create_resource(
+            {"ID": "urn:bar", "anchors": {"foo": 12}},
+        )
+        registry = Registry().with_resource(resource.id(), resource)
+        with pytest.raises(LookupError):
+            registry.anchor(resource.id(), "foo")
+
+        assert registry.crawl().anchor(resource.id(), "foo") == Anchor(
+            name="foo",
+            resource=ID_AND_CHILDREN.create_resource(12),
+        )
+
+    def test_crawl_finds_anchors_no_id(self):
+        resource = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}})
+        registry = Registry().with_resource("urn:root", resource)
+        with pytest.raises(LookupError):
+            registry.anchor("urn:root", "foo")
+
+        assert registry.crawl().anchor("urn:root", "foo") == Anchor(
+            name="foo",
+            resource=ID_AND_CHILDREN.create_resource(12),
+        )
+
     def test_contents(self):
         resource = Resource.opaque({"foo": "bar"})
         uri = "urn:example"
@@ -278,6 +309,7 @@ def test_id_delegates_to_specification(self):
             name="",
             id_of=lambda contents: "urn:fixedID",
             subresources_of=lambda contents: [],
+            anchors_in=lambda specification, contents: [],
         )
         resource = Resource(
             contents={"foo": "baz"},
@@ -298,6 +330,16 @@ def test_subresource_with_different_specification(self):
             DRAFT202012.create_resource(schema),
         ]
 
+    def test_anchors_delegates_to_specification(self):
+        resource = ID_AND_CHILDREN.create_resource(
+            {"anchors": {"foo": {}, "bar": 1, "baz": ""}},
+        )
+        assert list(resource.anchors()) == [
+            Anchor(name="foo", resource=ID_AND_CHILDREN.create_resource({})),
+            Anchor(name="bar", resource=ID_AND_CHILDREN.create_resource(1)),
+            Anchor(name="baz", resource=ID_AND_CHILDREN.create_resource("")),
+        ]
+
     def test_pointer_to_mapping(self):
         resource = Resource.opaque(contents={"foo": "baz"})
         resolver = Registry().resolver()
@@ -336,6 +378,23 @@ def test_lookup_subresource(self):
         resolved = resolver.lookup("http://example.com/a")
         assert resolved.contents == {"ID": "http://example.com/a", "foo": 12}
 
+    def test_lookup_anchor_with_id(self):
+        root = ID_AND_CHILDREN.create_resource(
+            {
+                "ID": "http://example.com/",
+                "anchors": {"foo": 12},
+            },
+        )
+        resolver = Registry().with_resource(root.id(), root).resolver()
+        resolved = resolver.lookup("http://example.com/#foo")
+        assert resolved.contents == 12
+
+    def test_lookup_anchor_without_id(self):
+        root = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}})
+        resolver = Registry().with_resource("urn:example", root).resolver()
+        resolved = resolver.lookup("urn:example#foo")
+        assert resolved.contents == 12
+
     def test_multiple_lookup(self):
         """
         Continuing to lookup resources maintains the new base URI.
@@ -382,6 +441,17 @@ def test_multiple_lookup_pointer(self):
         second = first.resolver.lookup("#/foo")
         assert second.contents == "bar"
 
+    def test_multiple_lookup_anchor(self):
+        root = ID_AND_CHILDREN.create_resource({"anchors": {"foo": 12}})
+        registry = Registry().with_resource("http://example.com/", root)
+
+        resolver = registry.resolver()
+        first = resolver.lookup("http://example.com/")
+        assert first.contents == {"anchors": {"foo": 12}}
+
+        second = first.resolver.lookup("#foo")
+        assert second.contents == 12
+
     def test_lookup_non_existent_pointer(self):
         resource = Resource.opaque({"foo": {}})
         resolver = Registry({"http://example.com/1": resource}).resolver()
@@ -412,6 +482,7 @@ def test_create_resource(self):
             name="",
             id_of=lambda contents: "urn:fixedID",
             subresources_of=lambda contents: [],
+            anchors_in=lambda specification, contents: [],
         )
         resource = specification.create_resource(contents={"foo": "baz"})
         assert resource == Resource(
@@ -446,6 +517,14 @@ def test_no_subresources(self, thing):
 
         assert list(Specification.OPAQUE.subresources_of(thing)) == []
 
+    @pytest.mark.parametrize("thing", THINGS)
+    def test_no_anchors(self, thing):
+        """
+        An arbitrary thing has no anchors.
+        """
+
+        assert list(Specification.OPAQUE.anchors_in(thing)) == []
+
 
 @pytest.mark.parametrize(
     "cls",