Add loaders.from_traversable for loading from within packages.

Author: Julian

noxfile.py

@@ -37,7 +37,11 @@ def tests(session):
     """
     Run the test suite with a corresponding Python version.
     """
-    session.install("-r", REQUIREMENTS["tests"])
+    session.install(
+        "-r",
+        REQUIREMENTS["tests"],
+        "importlib-resources; python_version<'3.11'",
+    )
 
     if session.posargs and session.posargs[0] == "coverage":
         if len(session.posargs) > 1 and session.posargs[1] == "github":

referencing_loaders/__init__.py

@@ -11,6 +11,8 @@
 from referencing import Resource, Specification
 
 if TYPE_CHECKING:
+    from importlib.resources.abc import Traversable
+
     from referencing.typing import URI
 
 
@@ -22,21 +24,55 @@ def from_path(root: Path) -> Iterable[tuple[URI, Resource[Any]]]:
     the root) -- though it still is often a good idea to explicitly indicate
     what specification every resource is written for internally.
     """
-    specification: Specification[Any] | None = None
-    for dir, _, files in _walk(root):
-        for file in files:
-            path = dir / file
-            contents = json.loads(path.read_text())
-            if specification is None:
-                specification = Specification.detect(contents)  # type: ignore[reportUnknownMemberType]
-            resource = specification.detect(contents).create_resource(contents)
-            yield path.as_uri(), resource
+    return _from_walked(_walk(root))
 
 
-def _walk(path: Path) -> Iterable[tuple[Path, Iterable[str], Iterable[str]]]:
+def _walk(path: Path) -> Iterable[Path]:
     walk = getattr(path, "walk", None)
-    if walk is not None:
-        yield from walk()
-        return
-    for root, dirs, files in os.walk(path):  # pragma: no cover
-        yield Path(root), dirs, files
+    if walk is None:
+        for dir, _, files in os.walk(path):  # pragma: no cover
+            for file in files:
+                yield Path(dir) / file
+    else:
+        for dir, _, files in walk():
+            for file in files:
+                yield dir / file
+
+
+def _walk_traversable(root: Traversable) -> Iterable[Traversable]:
+    """
+    .walk() for importlib resources paths, which don't have the method :/
+    """  # noqa: D415
+    walking = [root]
+    while walking:
+        path = walking.pop()
+        for each in path.iterdir():
+            if each.is_dir():
+                walking.append(each)
+            else:
+                yield each
+
+
+def from_traversable(root: Traversable) -> Iterable[tuple[URI, Resource[Any]]]:
+    """
+    Load some resources from a given `importlib.resources` traversable.
+
+    (I.e. load schemas from data within a Python package.)
+    """
+    return _from_walked(
+        each
+        for each in _walk_traversable(root)
+        if not each.name.endswith(".py")
+    )
+
+
+def _from_walked(
+    paths: Iterable[Path | Traversable],
+) -> Iterable[tuple[URI, Resource[Any]]]:
+    specification: Specification[Any] | None = None
+    for path in paths:
+        contents = json.loads(path.read_text())
+        if specification is None:
+            specification = Specification.detect(contents)  # type: ignore[reportUnknownMemberType]
+        resource = specification.detect(contents).create_resource(contents)
+        yield getattr(path, "as_uri", lambda: "")(), resource

referencing_loaders/tests/test_loaders.py

@@ -1,4 +1,10 @@
 import json
+import sys
+
+try:
+    from importlib.resources import files
+except ImportError:
+    from importlib_resources import files
 
 from referencing import Registry
 from referencing.jsonschema import DRAFT202012, EMPTY_REGISTRY
@@ -84,3 +90,37 @@ def test_schema_is_inherited_downwards(tmp_path):
 def test_empty(tmp_path):
     registry = EMPTY_REGISTRY.with_resources(loaders.from_path(tmp_path))
     assert registry == EMPTY_REGISTRY
+
+
+def test_traversable(tmp_path):
+    package = tmp_path / "foo"
+
+    schemas = package / "schemas"
+    path, schema = schemas / "schema.json", {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        "$id": "http://example.com/",
+    }
+
+    schemas.mkdir(parents=True)
+    package.joinpath("__init__.py").touch()
+    path.write_text(json.dumps(schema))
+
+    # ?!?! -- without this, importlib.resources.files fails on 3.9 and no other
+    # version!?!?
+    schemas.joinpath("__init__.py").touch()
+
+    try:
+        sys.path.append(str(tmp_path))
+        resources = loaders.from_traversable(files("foo.schemas"))
+        registry = EMPTY_REGISTRY.with_resources(resources)
+
+        assert (
+            registry.crawl()
+            == Registry()
+            .with_contents(
+                [(path.as_uri(), schema)],
+            )
+            .crawl()
+        )
+    finally:
+        sys.path.pop()