Define Registry.__rmatmul__ for adding resources.

I can't resist a bit of over-the-top fun. This is a (useful) shortcut
for adding a resource (or multiple resources) to a registry where the
URI you want to use is specifically the resource's internal ID.

It *must* have one to use this, otherwise you get an error, which is a
benefit over using `with_resource` -- which will happily not error if
you pass it `resource.id()` but that returned None.

That being said, Registries already have 3 methods named with_<foo> for
adding resource-like things to themselves, so I didn't want to add a
fourth and risk users getting even more confused as to which they should
use.

Using @ seems like the sort of thing that "pOwEr UsErS" will find, and
more casual users who just want to figure out something working will not
be distracted by.

Author: Julian

docs/api.rst

@@ -5,7 +5,7 @@ API Reference
    :members:
    :undoc-members:
    :imported-members:
-   :special-members: __iter__, __getitem__, __len__
+   :special-members: __iter__, __getitem__, __len__, __rmatmul__
 
 
 .. autoclass:: referencing._core.Resolved

referencing/_core.py

@@ -276,6 +276,45 @@ def __len__(self) -> int:
         """
         return len(self._resources)
 
+    def __rmatmul__(self, new: Resource[D] | Iterable[Resource[D]]):
+        """
+        Add resource(s) to a new registry, using their internal IDs.
+
+        Resources must have a internal IDs (e.g. the ``$id`` keyword in modern
+        JSON Schema versions), otherwise an error will be raised.
+
+        Use this via:
+
+            * ``resource @ registry`` or
+
+            * ``[iterable, of, multiple, resources] @ registry``
+
+        which -- again, assuming the resources have internal IDs -- is
+        equivalent to calling `Registry.with_resources` as such:
+
+        .. code:: python
+
+            registry.with_resources(
+                (resource.id(), resource) for resource in new_resources
+            )
+        """
+        if isinstance(new, Resource):
+            new = (new,)
+
+        resources = self._resources.evolver()
+        uncrawled = self._uncrawled.evolver()
+        for resource in new:
+            id = resource.id()
+            if id is None:
+                raise exceptions.NoInternalID(resource=resource)
+            uncrawled.add(id)
+            resources.set(id, resource)
+        return evolve(
+            self,
+            resources=resources.persistent(),
+            uncrawled=uncrawled.persistent(),
+        )
+
     def __repr__(self) -> str:
         size = len(self)
         pluralized = "resource" if size == 1 else "resources"

referencing/exceptions.py

@@ -1,13 +1,18 @@
 """
 Errors, oh no!
 """
-from typing import Any
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
 
 import attrs
 
 from referencing._attrs import frozen
 from referencing.typing import URI
 
+if TYPE_CHECKING:
+    from referencing import Resource
+
 
 @frozen
 class NoSuchResource(KeyError):
@@ -23,6 +28,25 @@ def __eq__(self, other: Any) -> bool:
         return attrs.astuple(self) == attrs.astuple(other)
 
 
+@frozen
+class NoInternalID(Exception):
+    """
+    A resource has no internal ID, but one is needed.
+
+    E.g. in modern JSON Schema drafts, this is the ``$id`` keyword.
+
+    One might be needed if a resource was to-be added to a registry but no
+    other URI is available, and the resource doesn't declare its canonical URI.
+    """
+
+    resource: Resource[Any]
+
+    def __eq__(self, other: Any) -> bool:
+        if self.__class__ is not other.__class__:
+            return NotImplemented
+        return attrs.astuple(self) == attrs.astuple(other)
+
+
 @frozen
 class Unretrievable(KeyError):
     """
@@ -64,7 +88,7 @@ class PointerToNowhere(Unresolvable):
     A JSON Pointer leads to a part of a document that does not exist.
     """
 
-    resource: Any
+    resource: Resource[Any]
 
     def __str__(self):
         return f"{self.ref!r} does not exist within {self.resource.contents!r}"
@@ -76,7 +100,7 @@ class NoSuchAnchor(Unresolvable):
     An anchor does not exist within a particular resource.
     """
 
-    resource: Any
+    resource: Resource[Any]
     anchor: str
 
     def __str__(self):

referencing/tests/test_core.py

@@ -56,6 +56,30 @@ def test_with_resources(self):
             resource=two,
         )
 
+    def test_matmul_resource(self):
+        uri = "urn:example:resource"
+        resource = ID_AND_CHILDREN.create_resource({"ID": uri, "foo": 12})
+        registry = resource @ Registry()
+        assert registry == Registry().with_resource(uri, resource)
+
+    def test_matmul_many_resources(self):
+        one_uri = "urn:example:one"
+        one = ID_AND_CHILDREN.create_resource({"ID": one_uri, "foo": 12})
+
+        two_uri = "urn:example:two"
+        two = ID_AND_CHILDREN.create_resource({"ID": two_uri, "foo": 12})
+
+        registry = [one, two] @ Registry()
+        assert registry == Registry().with_resources(
+            [(one_uri, one), (two_uri, two)],
+        )
+
+    def test_matmul_resource_without_id(self):
+        resource = Resource.opaque(contents={"foo": "bar"})
+        with pytest.raises(exceptions.NoInternalID) as e:
+            resource @ Registry()
+        assert e.value == exceptions.NoInternalID(resource=resource)
+
     def test_with_contents_from_json_schema(self):
         uri = "urn:example"
         schema = {"$schema": "https://json-schema.org/draft/2020-12/schema"}
@@ -98,7 +122,7 @@ def test_crawl_finds_a_subresource(self):
         root = ID_AND_CHILDREN.create_resource(
             {"ID": "urn:root", "children": [{"ID": child_id, "foo": 12}]},
         )
-        registry = Registry().with_resource(root.id(), root)
+        registry = root @ Registry()
         with pytest.raises(LookupError):
             registry[child_id]
 
@@ -109,7 +133,7 @@ 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)
+        registry = resource @ Registry()
 
         assert registry.crawl().anchor(resource.id(), "foo").value == Anchor(
             name="foo",
@@ -173,7 +197,7 @@ def test_dict_conversion(self):
         two = ID_AND_CHILDREN.create_resource({"foo": "bar"})
         registry = Registry(
             {"http://example.com/1": one},
-        ).with_resources([("http://example.com/foo/bar", two)])
+        ).with_resource("http://example.com/foo/bar", two)
         assert (
             registry.crawl()
             == Registry()
@@ -637,8 +661,8 @@ def test_lookup_subresource(self):
                 ],
             },
         )
-        resolver = Registry().with_resource(root.id(), root).resolver()
-        resolved = resolver.lookup("http://example.com/a")
+        registry = root @ Registry()
+        resolved = registry.resolver().lookup("http://example.com/a")
         assert resolved.contents == {"ID": "http://example.com/a", "foo": 12}
 
     def test_lookup_anchor_with_id(self):
@@ -648,8 +672,8 @@ def test_lookup_anchor_with_id(self):
                 "anchors": {"foo": 12},
             },
         )
-        resolver = Registry().with_resource(root.id(), root).resolver()
-        resolved = resolver.lookup("http://example.com/#foo")
+        registry = root @ Registry()
+        resolved = registry.resolver().lookup("http://example.com/#foo")
         assert resolved.contents == 12
 
     def test_lookup_anchor_without_id(self):
@@ -756,7 +780,7 @@ def test_in_subresource(self):
                 ],
             },
         )
-        registry = Registry().with_resource(root.id(), root)
+        registry = root @ Registry()
 
         resolver = registry.resolver()
         first = resolver.lookup("http://example.com/")
@@ -783,7 +807,7 @@ def test_in_pointer_subresource(self):
                 ],
             },
         )
-        registry = Registry().with_resource(root.id(), root)
+        registry = root @ Registry()
 
         resolver = registry.resolver()
         first = resolver.lookup("http://example.com/")
@@ -814,9 +838,7 @@ def test_dynamic_scope(self):
                 "children": [{"ID": "two-child/"}],
             },
         )
-        registry = Registry().with_resources(
-            [(one.id(), one), (two.id(), two)],
-        )
+        registry = [one, two] @ Registry()
 
         resolver = registry.resolver()
         first = resolver.lookup("http://example.com/")