add custom validator as input (#1320)

* add custom validator as input

* update changelog

* update validator arg in docstrings

* add unittest for new custom validator argument

* fix linting issues

Author: fmigneault

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Changed
 
+- Add `validator` input to `STACObject.validate` for inline reference of the validator to use
+  ([#1320](https://github.com/stac-utils/pystac/pull/1320))
 - Made item pickles smaller by changing how nested links are stored([#1285](https://github.com/stac-utils/pystac/pull/1285))
 - Add APILayoutStrategy ([#1294](https://github.com/stac-utils/pystac/pull/1294))
 - Allow setting a default layout strategy for Catalog ([#1295](https://github.com/stac-utils/pystac/pull/1295))

pystac/stac_object.py

@@ -61,19 +61,27 @@ def __init__(self, stac_extensions: list[str]) -> None:
         self.links = []
         self.stac_extensions = stac_extensions
 
-    def validate(self) -> list[Any]:
+    def validate(
+        self,
+        validator: pystac.validation.stac_validator.STACValidator | None = None,
+    ) -> list[Any]:
         """Validate this STACObject.
 
         Returns a list of validation results, which depends on the validation
-        implementation. For JSON Schema validation, this will be a list
-        of schema URIs that were used during validation.
+        implementation. For JSON Schema validation (default validator), this
+        will be a list of schema URIs that were used during validation.
 
+        Args:
+            validator : A custom validator to use for validation of the object.
+                If omitted, the default validator from
+                :class:`~pystac.validation.RegisteredValidator`
+                will be used instead.
         Raises:
             STACValidationError
         """
         import pystac.validation
 
-        return pystac.validation.validate(self)
+        return pystac.validation.validate(self, validator=validator)
 
     def add_link(self, link: Link) -> None:
         """Add a link to this object's set of links.

pystac/validation/__init__.py

@@ -19,11 +19,18 @@
 from pystac.validation.stac_validator import JsonSchemaSTACValidator, STACValidator
 
 
-def validate(stac_object: STACObject) -> list[Any]:
+def validate(
+    stac_object: STACObject,
+    validator: STACValidator | None = None,
+) -> list[Any]:
     """Validates a :class:`~pystac.STACObject`.
 
     Args:
         stac_object : The stac object to validate.
+        validator : A custom validator to use for validation of the STAC object.
+            If omitted, the default validator from
+            :class:`~pystac.validation.RegisteredValidator`
+            will be used instead.
 
     Returns:
         List[Object]: List of return values from the validation calls for the
@@ -39,6 +46,7 @@ def validate(stac_object: STACObject) -> list[Any]:
         stac_version=pystac.get_stac_version(),
         extensions=stac_object.stac_extensions,
         href=stac_object.get_self_href(),
+        validator=validator,
     )
 
 
@@ -48,6 +56,7 @@ def validate_dict(
     stac_version: str | None = None,
     extensions: list[str] | None = None,
     href: str | None = None,
+    validator: STACValidator | None = None,
 ) -> list[Any]:
     """Validate a stac object serialized as JSON into a dict.
 
@@ -67,6 +76,10 @@ def validate_dict(
         extensions : Extension IDs for this stac object. If not supplied,
             PySTAC's identification logic to identify the extensions.
         href : Optional HREF of the STAC object being validated.
+        validator : A custom validator to use for validation of the STAC dictionary.
+            If omitted, the default validator from
+            :class:`~pystac.validation.RegisteredValidator`
+            will be used instead.
 
     Returns:
         List[Object]: List of return values from the validation calls for the
@@ -104,7 +117,8 @@ def _get_uri(ext: str) -> str | None:
 
         extensions = [uri for uri in map(_get_uri, extensions) if uri is not None]
 
-    return RegisteredValidator.get_validator().validate(
+    validator = validator or RegisteredValidator.get_validator()
+    return validator.validate(
         stac_dict, stac_object_type, stac_version, extensions, href
     )
 
@@ -248,4 +262,8 @@ def set_validator(validator: STACValidator) -> None:
     RegisteredValidator.set_validator(validator)
 
 
-__all__ = ["GetSchemaError"]
+__all__ = [
+    "GetSchemaError",
+    "JsonSchemaSTACValidator",
+    "RegisteredValidator",
+]

tests/validation/test_validate.py

@@ -3,7 +3,7 @@
 import shutil
 import tempfile
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, cast
 
 import jsonschema
 import pytest
@@ -14,7 +14,7 @@
 from pystac.cache import CollectionCache
 from pystac.serialization.common_properties import merge_common_properties
 from pystac.utils import get_opt
-from pystac.validation import GetSchemaError
+from pystac.validation import GetSchemaError, JsonSchemaSTACValidator
 from tests.utils import TestCases
 from tests.utils.test_cases import ExampleInfo
 
@@ -183,6 +183,59 @@ def test_validates_geojson_with_tuple_coordinates(self) -> None:
         # Should not raise.
         item.validate()
 
+    @pytest.mark.vcr()
+    def test_validate_custom_validator(self) -> None:
+        """This test verifies the use of a custom validator class passed as
+        input to :meth:`~pystac.stac_object.STACObject.validate` and every
+        underlying function. This validator is effective only for the call
+        for which it was provided, contrary to
+        :class:`~pystac.validation.RegisteredValidator`
+        that persists it globally until reset.
+        """
+        custom_extension_uri = (
+            "https://stac-extensions.github.io/custom-extension/v1.0.0/schema.json"
+        )
+        custom_extension_schema = {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "$id": f"{custom_extension_uri}#",
+            "type": "object",
+            "properties": {
+                "properties": {
+                    "type": "object",
+                    "required": ["custom-extension:test"],
+                    "properties": {"custom-extension:test": {"type": "integer"}},
+                }
+            },
+        }
+        item = cast(
+            pystac.Item,
+            pystac.read_file(TestCases.get_path("data-files/item/sample-item.json")),
+        )
+        item.stac_extensions.append(custom_extension_uri)
+        item.properties["custom-extension:test"] = 123
+
+        with pytest.raises(pystac.validation.GetSchemaError):
+            item.validate()  # default validator does not know the extension
+
+        class CustomValidator(JsonSchemaSTACValidator):
+            def _get_schema(self, schema_uri: str) -> dict[str, Any]:
+                if schema_uri == custom_extension_uri:
+                    return custom_extension_schema
+                return super()._get_schema(schema_uri)
+
+        # validate that the custom schema is found with the custom validator
+        custom_validator = CustomValidator()
+        item.validate(validator=custom_validator)
+
+        # make sure validation is effective
+        item.properties["custom-extension:test"] = "bad-value"
+        with pytest.raises(pystac.errors.STACValidationError):
+            item.validate(validator=custom_validator)
+
+        # verify that the custom validator is not persisted
+        with pytest.raises(pystac.validation.GetSchemaError):
+            item.validate()
+
 
 @pytest.mark.block_network
 def test_catalog_latest_version_uses_local(catalog: pystac.Catalog) -> None: