Add stac-pydantic validation

.github/workflows/test-runner.yml

@@ -32,7 +32,7 @@ jobs:
           pytest --mypy stac_validator
 
       - name: Run pre-commit
-        if: matrix.python-version == 3.10
+        if: matrix.python-version == 3.12
         run: |
           pre-commit install
           pre-commit autoupdate

CHANGELOG.md

@@ -10,6 +10,8 @@ The format is (loosely) based on [Keep a Changelog](http://keepachangelog.com/)
 
 - Added a comprehensive Table of Contents to README.md [#247](https://github.com/stac-utils/stac-validator/pull/247)
 - Added Sponsors and Supporters section to README.md with organizational logos and acknowledgments [#247](https://github.com/stac-utils/stac-validator/pull/247)
+- Added `--pydantic` option for validating STAC objects using stac-pydantic models, providing enhanced type checking and validation [#249](https://github.com/stac-utils/stac-validator/pull/249)
+- Added optional dependency for stac-pydantic that can be installed with `pip install stac-validator[pydantic]` [#249](https://github.com/stac-utils/stac-validator/pull/249)
 
 ### Changed
 

README.md

@@ -31,6 +31,7 @@
   - [Item Collection Validation](#--item-collection)
   - [Using Headers](#--header)
   - [Schema Mapping](#--schema-map)
+  - [Pydantic Validation](#--pydantic)
 - [Deployment](#deployment)
   - [Docker](#docker)
   - [AWS (CDK)](#aws-cdk)
@@ -174,6 +175,8 @@ Options:
   --no_output                     Do not print output to console.
   --log_file TEXT                 Save full recursive output to log file
                                   (local filepath).
+  --pydantic                      Validate using stac-pydantic models for enhanced
+                                  type checking and validation.
   --help                          Show this message and exit.
 ```
 
@@ -458,6 +461,43 @@ $ stac-validator tests/test_data/v100/extended-item-local.json --custom tests/te
 ]
 ```
 
+### --pydantic
+
+The `--pydantic` option provides enhanced validation using stac-pydantic models, which offer stronger type checking and more detailed error messages. To use this feature, you need to install the optional dependency:
+
+```bash
+$ pip install stac-validator[pydantic]
+```
+
+Then you can validate your STAC objects using Pydantic models:
+
+```bash
+$ stac-validator https://raw.githubusercontent.com/radiantearth/stac-spec/master/examples/extended-item.json --pydantic
+```
+
+```bash
+[
+    {
+        "version": "1.0.0",
+        "path": "https://raw.githubusercontent.com/radiantearth/stac-spec/master/examples/extended-item.json",
+        "schema": [
+            "stac-pydantic Item model"
+        ],
+        "valid_stac": true,
+        "asset_type": "ITEM",
+        "validation_method": "pydantic",
+        "extension_schemas": [
+            "https://stac-extensions.github.io/eo/v1.0.0/schema.json",
+            "https://stac-extensions.github.io/projection/v1.0.0/schema.json",
+            "https://stac-extensions.github.io/scientific/v1.0.0/schema.json",
+            "https://stac-extensions.github.io/view/v1.0.0/schema.json",
+            "https://stac-extensions.github.io/remote-data/v1.0.0/schema.json"
+        ],
+        "model_validation": "passed"
+    }
+]
+```
+
 ## Sponsors and Supporters
 
 The following organizations have contributed time and/or funding to support the development of this project:

setup.py

@@ -29,6 +29,7 @@
         "requests>=2.32.3",
         "jsonschema>=4.23.0",
         "click>=8.1.8",
+        "stac-pydantic>=3.3.0",
         "referencing>=0.35.1",
     ],
     extras_require={
@@ -37,6 +38,9 @@
             "requests-mock",
             "types-setuptools",
         ],
+        "pydantic": [
+            "stac-pydantic>=3.3.0",
+        ],
     },
     packages=["stac_validator"],
     entry_points={

stac_validator/stac_validator.py

@@ -142,6 +142,11 @@ def collections_summary(message: List[Dict[str, Any]]) -> None:
     default="",
     help="Save full recursive output to log file (local filepath).",
 )
+@click.option(
+    "--pydantic",
+    is_flag=True,
+    help="Validate using stac-pydantic models for enhanced type checking and validation.",
+)
 def main(
     stac_file: str,
     collections: bool,
@@ -160,6 +165,7 @@ def main(
     verbose: bool,
     no_output: bool,
     log_file: str,
+    pydantic: bool,
 ) -> None:
     """Main function for the `stac-validator` command line tool. Validates a STAC file
     against the STAC specification and prints the validation results to the console as JSON.
@@ -182,6 +188,7 @@ def main(
         verbose (bool): Whether to enable verbose output for recursive mode.
         no_output (bool): Whether to print output to console.
         log_file (str): Path to a log file to save full recursive output.
+        pydantic (bool): Whether to validate using stac-pydantic models for enhanced type checking and validation.
 
     Returns:
         None
@@ -212,6 +219,7 @@ def main(
         schema_map=schema_map_dict,
         verbose=verbose,
         log=log_file,
+        pydantic=pydantic,
     )
     if not item_collection and not collections:
         valid = stac.run()

stac_validator/validate.py

@@ -40,6 +40,7 @@ class StacValidate:
         custom (str): The local filepath or remote URL of a custom JSON schema to validate the STAC object.
         verbose (bool): Whether to enable verbose output in recursive mode.
         log (str): The local filepath to save the output of the recursive validation to.
+        pydantic (bool): Whether to validate using Pydantic models.
 
     Methods:
         run(): Validates the STAC object and returns whether it is valid.
@@ -64,6 +65,7 @@ def __init__(
         schema_map: Optional[Dict[str, str]] = None,
         verbose: bool = False,
         log: str = "",
+        pydantic: bool = False,
     ):
         self.stac_file = stac_file
         self.collections = collections
@@ -87,6 +89,7 @@ def __init__(
         self.verbose = verbose
         self.valid = False
         self.log = log
+        self.pydantic = pydantic
 
     @property
     def schema(self) -> str:
@@ -372,7 +375,7 @@ def recursive_validator(self, stac_type: str) -> bool:
                 if e.absolute_path:
                     err_msg = (
                         f"{e.message}. Error is in "
-                        f"{' -> '.join([str(i) for i in e.absolute_path])}"
+                        f"{' -> '.join([str(i) for i in e.absolute_path])} "
                     )
                 else:
                     err_msg = f"{e.message}"
@@ -474,8 +477,9 @@ def validate_collections(self) -> None:
 
         Raises:
             URLError, JSONDecodeError, ValueError, TypeError, FileNotFoundError,
-            ConnectionError, exceptions.SSLError, OSError: Various errors related
-            to fetching or parsing.
+            ConnectionError, exceptions.SSLError, OSError, KeyError, HTTPError,
+            jsonschema.exceptions.ValidationError, Exception: Various errors
+            during fetching or parsing.
         """
         collections = fetch_and_parse_file(str(self.stac_file), self.headers)
         for collection in collections["collections"]:
@@ -488,8 +492,9 @@ def validate_item_collection(self) -> None:
 
         Raises:
             URLError, JSONDecodeError, ValueError, TypeError, FileNotFoundError,
-            ConnectionError, exceptions.SSLError, OSError: Various errors related
-            to fetching or parsing.
+            ConnectionError, exceptions.SSLError, OSError, KeyError, HTTPError,
+            jsonschema.exceptions.ValidationError, Exception: Various errors
+            during fetching or parsing.
         """
         page = 1
         print(f"processing page {page}")
@@ -519,6 +524,88 @@ def validate_item_collection(self) -> None:
             }
             self.message.append(message)
 
+    def pydantic_validator(self, stac_type: str) -> Dict:
+        """
+        Validate STAC content using Pydantic models.
+
+        Args:
+            stac_type (str): The STAC object type (e.g., "ITEM", "COLLECTION", "CATALOG").
+
+        Returns:
+            dict: A dictionary containing validation results.
+        """
+        message = self.create_message(stac_type, "pydantic")
+        message["schema"] = [""]
+
+        try:
+            # Import dependencies
+            from pydantic import ValidationError  # type: ignore
+            from stac_pydantic import Catalog, Collection, Item  # type: ignore
+            from stac_pydantic.extensions import validate_extensions  # type: ignore
+
+            # Validate based on STAC type
+            if stac_type == "ITEM":
+                item_model = Item.model_validate(self.stac_content)
+                message["schema"] = ["stac-pydantic Item model"]
+                self._validate_extensions(item_model, message, validate_extensions)
+
+            elif stac_type == "COLLECTION":
+                collection_model = Collection.model_validate(self.stac_content)
+                message["schema"] = [
+                    "stac-pydantic Collection model"
+                ]  # Fix applied here
+                self._validate_extensions(
+                    collection_model, message, validate_extensions
+                )
+
+            elif stac_type == "CATALOG":
+                Catalog.model_validate(self.stac_content)
+                message["schema"] = ["stac-pydantic Catalog model"]
+
+            else:
+                raise ValueError(
+                    f"Unsupported STAC type for Pydantic validation: {stac_type}"
+                )
+
+            self.valid = True
+            message["model_validation"] = "passed"
+
+        except ValidationError as e:
+            self.valid = False
+            error_details = [
+                f"{' -> '.join(map(str, error.get('loc', [])))}: {error.get('msg', '')}"
+                for error in e.errors()
+            ]
+            error_message = f"Pydantic validation failed for {stac_type}: {'; '.join(error_details)}"
+            message.update(
+                self.create_err_msg("PydanticValidationError", error_message)
+            )
+
+        except Exception as e:
+            self.valid = False
+            message.update(self.create_err_msg("PydanticValidationError", str(e)))
+
+        return message
+
+    def _validate_extensions(self, model, message: Dict, validate_extensions) -> None:
+        """
+        Validate extensions for a given Pydantic model.
+
+        Args:
+            model: The Pydantic model instance.
+            message (dict): The validation message dictionary to update.
+            validate_extensions: The function to validate extensions.
+        """
+        if (
+            "stac_extensions" in self.stac_content
+            and self.stac_content["stac_extensions"]
+        ):
+            extension_schemas = []
+            validate_extensions(model, reraise_exception=True)
+            for ext in self.stac_content["stac_extensions"]:
+                extension_schemas.append(ext)
+            message["extension_schemas"] = extension_schemas
+
     def run(self) -> bool:
         """
         Run the STAC validation process based on the input parameters.
@@ -563,6 +650,9 @@ def run(self) -> bool:
             elif self.extensions:
                 message = self.extensions_validator(stac_type)
 
+            elif self.pydantic:
+                message = self.pydantic_validator(stac_type)
+
             else:
                 self.valid = True
                 message = self.default_validator(stac_type)

tests/test_pydantic.py

@@ -0,0 +1,61 @@
+"""
+Description: Test the validator for Pydantic model validation
+
+"""
+
+from stac_validator import stac_validator
+
+
+def test_pydantic_item_local_v110():
+    stac_file = "tests/test_data/v110/simple-item.json"
+    stac = stac_validator.StacValidate(stac_file, pydantic=True)
+    stac.run()
+    assert stac.message[0]["version"] == "1.1.0"
+    assert stac.message[0]["path"] == "tests/test_data/v110/simple-item.json"
+    assert stac.message[0]["schema"] == ["stac-pydantic Item model"]
+    assert stac.message[0]["valid_stac"] is True
+    assert stac.message[0]["asset_type"] == "ITEM"
+    assert stac.message[0]["validation_method"] == "pydantic"
+    assert stac.message[0]["model_validation"] == "passed"
+
+
+def test_pydantic_collection_local_v110():
+    stac_file = "tests/test_data/v110/collection.json"
+    stac = stac_validator.StacValidate(stac_file, pydantic=True)
+    stac.run()
+    assert stac.message[0]["version"] == "1.1.0"
+    assert stac.message[0]["path"] == "tests/test_data/v110/collection.json"
+    assert stac.message[0]["schema"] == ["stac-pydantic Collection model"]
+    assert stac.message[0]["valid_stac"] is True
+    assert stac.message[0]["asset_type"] == "COLLECTION"
+    assert stac.message[0]["validation_method"] == "pydantic"
+    assert "extension_schemas" in stac.message[0]
+    assert stac.message[0]["model_validation"] == "passed"
+
+
+# Find a catalog file in v100 directory since v110 doesn't have one
+def test_pydantic_catalog_local_v100():
+    stac_file = "tests/test_data/v100/catalog.json"
+    stac = stac_validator.StacValidate(stac_file, pydantic=True)
+    stac.run()
+    assert stac.message[0]["version"] == "1.0.0"
+    assert stac.message[0]["path"] == "tests/test_data/v100/catalog.json"
+    assert stac.message[0]["schema"] == ["stac-pydantic Catalog model"]
+    assert stac.message[0]["valid_stac"] is True
+    assert stac.message[0]["asset_type"] == "CATALOG"
+    assert stac.message[0]["validation_method"] == "pydantic"
+    assert stac.message[0]["model_validation"] == "passed"
+
+
+def test_pydantic_invalid_item():
+    # Test with a file that should fail Pydantic validation
+    stac_file = "tests/test_data/bad_data/bad_item_v090.json"
+    stac = stac_validator.StacValidate(stac_file, pydantic=True)
+    stac.run()
+    assert stac.message[0]["version"] == "0.9.0"
+    assert stac.message[0]["path"] == "tests/test_data/bad_data/bad_item_v090.json"
+    assert stac.message[0]["valid_stac"] is False
+    assert stac.message[0]["asset_type"] == "ITEM"
+    assert stac.message[0]["validation_method"] == "pydantic"
+    assert "error_type" in stac.message[0]
+    assert "error_message" in stac.message[0]

tests/test_recursion.py

@@ -398,6 +398,6 @@ def test_recursion_with_missing_collection_link():
             "valid_stac": False,
             "validation_method": "recursive",
             "error_type": "JSONSchemaValidationError",
-            "error_message": "'simple-collection' should not be valid under {}. Error is in collection",
+            "error_message": "'simple-collection' should not be valid under {}. Error is in collection ",
         },
     ]

tox.ini

@@ -5,4 +5,5 @@ envlist = py38,py39,py310,py311,py312,py313
 deps = 
     pytest
     requests-mock
-commands = pytest
+commands = pytest
+extras = pydantic