Add a way to dynamically retrieve resources in a Registry.

Allows for e.g. hooking up an HTTP client to retrieve resources from
the network, the filesystem, a database, etc.

Of course none of this happens by default.

Author: Julian

referencing/_core.py

@@ -174,6 +174,10 @@ def pointer(self, pointer: str, resolver: Resolver[D]) -> Resolved[D]:
         return Resolved(contents=contents, resolver=resolver)  # type: ignore[reportUnknownArgumentType]  # noqa: E501
 
 
+def _fail_to_retrieve(uri: URI):
+    raise exceptions.NoSuchResource(ref=uri)
+
+
 @frozen
 class Registry(Mapping[URI, Resource[D]]):
     r"""
@@ -190,17 +194,32 @@ class Registry(Mapping[URI, Resource[D]]):
 
     Registries are immutable, and their methods return new instances of the
     registry with the additional resources added to them.
+
+    The ``retrieve`` argument can be used to configure retrieval of resources
+    dynamically, either over the network, from a database, or the like.
+    Pass it a callable which will be called if any URI not present in the
+    registry is accessed. It must either return a `Resource` or else raise an
+    exception indicating that the resource is not retrievable.
     """
 
     _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
+    _retrieve: Callable[[URI], Resource[D]] = field(default=_fail_to_retrieve)
 
     def __getitem__(self, uri: URI) -> Resource[D]:
         """
         Return the `Resource` identified by the given URI.
         """
-        return self._resources[uri]
+        try:
+            return self._resources[uri]
+        except LookupError:
+            try:
+                return self._retrieve(uri)
+            except exceptions.NoSuchResource:
+                raise
+            except Exception:
+                raise exceptions.NoSuchResource(ref=uri)
 
     def __iter__(self) -> Iterator[URI]:
         """
@@ -393,7 +412,7 @@ def lookup(self, ref: URI) -> Resolved[D]:
             registry = registry.crawl()
             try:
                 resource = registry[uri]
-            except KeyError:
+            except exceptions.NoSuchResource:
                 raise exceptions.Unresolvable(ref=ref) from None
 
         if fragment.startswith("/"):

referencing/exceptions.py

@@ -9,6 +9,15 @@
 from referencing.typing import URI
 
 
+@frozen
+class NoSuchResource(KeyError):
+    """
+    The given URI is not present in a registry.
+    """
+
+    ref: URI
+
+
 @frozen
 class CannotDetermineSpecification(Exception):
     """

referencing/tests/test_core.py

@@ -375,6 +375,39 @@ def test_opaque(self):
             specification=Specification.OPAQUE,
         )
 
+    def test_retrieve(self):
+        foo = Resource.opaque({"foo": "bar"})
+        registry = Registry(retrieve=lambda uri: foo)
+        assert registry["urn:example"] == foo
+
+    def test_retrieve_error(self):
+        def retrieve(uri):
+            if uri == "urn:succeed":
+                return {}
+            raise Exception("Oh no!")
+
+        registry = Registry(retrieve=retrieve)
+        assert registry["urn:succeed"] == {}
+        with pytest.raises(exceptions.NoSuchResource):
+            registry["urn:uhoh"]
+
+    def test_retrieve_already_available_resource(self):
+        def retrieve(uri):
+            raise Exception("Oh no!")
+
+        foo = Resource.opaque({"foo": "bar"})
+        registry = Registry({"urn:example": foo})
+        assert registry["urn:example"] == foo
+
+    def test_retrieve_crawlable_resource(self):
+        def retrieve(uri):
+            raise Exception("Oh no!")
+
+        child = ID_AND_CHILDREN.create_resource({"ID": "urn:child", "foo": 12})
+        root = ID_AND_CHILDREN.create_resource({"children": [child.contents]})
+        registry = Registry(retrieve=retrieve).with_resource("urn:root", root)
+        assert registry.crawl()["urn:child"] == child
+
 
 class TestResolver:
     def test_lookup_exact_uri(self):