Add dynamic fetching of remote schemas

pystac/validation/__init__.py

@@ -4,11 +4,12 @@
 
 import pystac
 from pystac.serialization.identify import STACVersionID, identify_stac_object
+from pystac.stac_object import STACObjectType
 from pystac.utils import make_absolute_href
 from pystac.validation.schema_uri_map import OldExtensionSchemaUriMap
 
 if TYPE_CHECKING:
-    from pystac.stac_object import STACObject, STACObjectType
+    from pystac.stac_object import STACObject
 
 
 # Import after above class definition

pystac/validation/stac_validator.py

@@ -10,10 +10,13 @@
 from pystac.validation.schema_uri_map import DefaultSchemaUriMap, SchemaUriMap
 
 try:
+    from urllib.request import Request, urlopen
+
     import jsonschema
     import jsonschema.exceptions
     import jsonschema.validators
     from referencing import Registry, Resource
+    from referencing.typing import D
 
     from pystac.validation.local_validator import LocalValidator
 
@@ -158,7 +161,7 @@ def get_schema_from_uri(self, schema_uri: str) -> Tuple[Dict[str, Any], Any]:
             self.schema_cache[schema_uri] = s
 
         schema = self.schema_cache[schema_uri]
-        registry = Registry().with_resources(
+        registry = Registry(retrieve=_retrieve).with_resources(  # type: ignore
             [
                 (k, Resource.from_contents(v)) for k, v in self.schema_cache.items()
             ]  # type: ignore
@@ -279,3 +282,10 @@ def validate_extension(
         self._validate_from_uri(stac_dict, stac_object_type, schema_uri, href)
 
         return schema_uri
+
+
+def _retrieve(href: str) -> Resource[D]:
+    request = Request(href)
+    with urlopen(request) as f:
+        response = f.read().decode("utf-8")
+    return Resource.from_contents(json.loads(response))