Support Pydantic Models for Validation (#165)

---------

Co-authored-by: Christian Adell <chadell@gmail.com>

Author: FragmentedPacket

docs/custom_validators.md

@@ -3,8 +3,7 @@
 With custom validators, you can implement business logic in Python. Schema-enforcer will automatically
 load your plugins from the `validator_directory` and run them against your host data.
 
-The validator plugin provides two base classes: ModelValidation and JmesPathModelValidation. The former can be used
-when you want to implement all logic and the latter can be used as a shortcut for jmespath validation.
+The validator plugin provides a few base classes: BaseValidation, JmesPathModelValidation, and PydanticValidation. BaseValidation can be used when you want to implement all logic, JmesPathModelValidation can be used as a shortcut for jmespath validation, and PydanticValidation will validate data against a specific Pydantic model.
 
 ## BaseValidation
 
@@ -26,7 +25,7 @@ by providing a class-level `id` variable.
 
 Helper functions are provided to add pass/fail results:
 
-```
+```python
 def add_validation_error(self, message: str, **kwargs):
     """Add validator error to results.
     Args:
@@ -40,6 +39,7 @@ def add_validation_pass(self, **kwargs):
       kwargs (optional): additional arguments to add to ValidationResult when required
     """
 ```
+
 In most cases, you will not need to provide kwargs. However, if you find a use case that requires updating other fields
 in the ValidationResult, you can send the key/value pairs to update the result directly. This is for advanced users only.
 
@@ -59,7 +59,7 @@ the following criteria:
    * `operator`: Operator to use for comparison between left and right hand side of expression
    * `error`: Message to report when validation fails
 
-### Supported operators:
+### Supported operators
 
 The class provides the following operators for basic use cases:
 
@@ -74,10 +74,11 @@ The class provides the following operators for basic use cases:
 
 If you require additional logic or need to compare other types, use the BaseValidation class and create your own validate method.
 
-### Examples:
+### Examples
 
 #### Basic
-```
+
+```python
 from schema_enforcer.schemas.validator import JmesPathModelValidation
 
 class CheckInterface(JmesPathModelValidation):  # pylint: disable=too-few-public-methods
@@ -90,7 +91,8 @@ class CheckInterface(JmesPathModelValidation):  # pylint: disable=too-few-public
 ```
 
 #### With compiled jmespath expression
-```
+
+```python
 import jmespath
 from schema_enforcer.schemas.validator import JmesPathModelValidation
 
@@ -104,6 +106,100 @@ class CheckInterfaceIPv4(JmesPathModelValidation):  # pylint: disable=too-few-pu
     error = "All core interfaces do not have IPv4 addresses"
 ```
 
+## PydanticValidation
+
+Schema Enforcer supports utilizing Pydantic models for validation. Pydantic models can be loaded two different ways.
+
+1. Store your models in your `validator_directory`.
+2. Load from a separate library using the `schema_enforcer.schemas.PydanticManager`. These must be defined within the `schema_enforcer` configuration file.
+   1. `pydantic_validators` requires a list of library paths to a `PydanticManager` instance.
+
+Both methods will replace the Pydantic `BaseModel` with the `PydanticValidation` class that provides the required `validate` method that uses the `model_validate` method to validate data. The model is set to the original Pydantic model to validate data against.
+
+### Pydantic Models in External Libraries
+
+```python
+class PydanticValidation(BaseValidation):
+    """Basic wrapper for Pydantic models to be used as validators."""
+
+    model: BaseModel
+
+    def validate(self, data: dict, strict: bool = False):
+        """Validate data against Pydantic model.
+
+        Args:
+            data (dict): variables to be validated by validator
+            strict (bool): true when --strict cli option is used to request strict validation (if provided)
+
+        Returns:
+            None
+
+        Use add_validation_error and add_validation_pass to report results.
+        """
+        try:
+            self.model.model_validate(data, strict=strict)
+            self.add_validation_pass()
+        except ValidationError as err:
+            self.add_validation_error(str(err))
+```
+
+### Pydantic Models in Validators Directory
+
+The Pydantic models can be located in any Python file within this directory (new or existing). The only requirement is these are valid Pydantic `BaseModel` subclasses.
+
+These will be loaded and can be referenced by their class name. For example, `CheckHostname` will show up as `CheckHostname`.
+
+```python
+"""Validate hostname is valid."""
+from pydantic import BaseModel
+
+
+class CheckHostname(BaseModel):
+    """Validate hostname is valid."""
+
+    hostname: str
+```
+
+```yaml
+# jsonschema: Hostname
+---
+hostname: "az-phx-rtr01"
+```
+
+### Load from External Library
+
+As an example, we will look at models that are within our `my_custom_pydantic_models.manager`. If a **prefix** is defined, you can reference the validators like `f"{prefix}/{model.__name__}`.
+
+```python
+"""Load our models to be used for Schema Enforcer."""
+from pydantic import BaseModel
+from schema_enforcer.schemas.manager import PydanticManager
+
+
+class Hostname(BaseModel):
+    hostname: str = Field(pattern="^[a-z]{2}-[a-z]{3}-[a-z]{1,2}[0-9]{2}$")
+
+
+# Prefix is optional and will default to a blank string, aka no prefix.
+# Models is required to pass in custom Pydantic models.
+manager = PydanticManager(prefix="custom", models=[Hostname])
+```
+
+```toml
+[tool.schema_enforcer]
+pydantic_validators = [
+  "my_custom_pydantic_models.manager"
+]
+```
+
+An example YAML file schema correlation would look like:
+
+```yaml
+# jsonschema: custom/Hostname
+---
+hostname: "az-phx-pe01"
+```
+
 ## Running validators
 
 Custom validators are run with `schema-enforcer validate` and `schema-enforcer ansible` commands.

pyproject.toml

@@ -63,6 +63,7 @@ exclude = '''
         | dist
         )/
     '''
+
 [tool.pylint.master]
 ignore=".venv"
 

schema_enforcer/cli.py

@@ -23,7 +23,13 @@ def main():
     """
 
 
-@click.option("--show-pass", default=False, help="Shows validation checks that passed", is_flag=True, show_default=True)
+@click.option(
+    "--show-pass",
+    default=False,
+    help="Shows validation checks that passed",
+    is_flag=True,
+    show_default=True,
+)
 @click.option(
     "--strict",
     default=False,
@@ -135,7 +141,11 @@ def validate(show_pass, show_checks, strict):  # noqa D205
     is_flag=True,
 )
 @click.option(
-    "--schema-id", default=None, cls=MutuallyExclusiveOption, mutually_exclusive=["list"], help="The name of a schema."
+    "--schema-id",
+    default=None,
+    cls=MutuallyExclusiveOption,
+    mutually_exclusive=["list"],
+    help="The name of a schema.",
 )
 @main.command()
 def schema(check, generate_invalid, list_schemas, schema_id, dump_schemas):  # noqa: D417,D301,D205
@@ -192,8 +202,20 @@ def schema(check, generate_invalid, list_schemas, schema_id, dump_schemas):  # n
 
 @main.command()
 @click.option("--inventory", "-i", help="Ansible inventory file.", required=False)
-@click.option("--host", "-h", "limit", help="Limit the execution to a single host.", required=False)
-@click.option("--show-pass", default=False, help="Shows validation checks that passed", is_flag=True, show_default=True)
+@click.option(
+    "--host",
+    "-h",
+    "limit",
+    help="Limit the execution to a single host.",
+    required=False,
+)
+@click.option(
+    "--show-pass",
+    default=False,
+    help="Shows validation checks that passed",
+    is_flag=True,
+    show_default=True,
+)
 @click.option(
     "--show-checks",
     default=False,

schema_enforcer/config.py

@@ -4,6 +4,7 @@
 import sys
 from pathlib import Path
 from typing import Dict, List, Optional
+from typing_extensions import Annotated
 
 import toml
 from pydantic import Field, ValidationError
@@ -31,6 +32,7 @@ class Settings(BaseSettings):  # pylint: disable=too-few-public-methods
     definition_directory: str = "definitions"
     schema_directory: str = "schemas"
     validator_directory: str = "validators"
+    pydantic_validators: Optional[List[Annotated[str, Field(pattern="^.*:.*$")]]] = Field(default_factory=list)
     test_directory: str = "tests"
 
     # Settings specific to the schema files

schema_enforcer/instances/file.py

@@ -3,6 +3,7 @@
 import re
 import itertools
 from pathlib import Path
+from ruamel.yaml.comments import CommentedMap
 from schema_enforcer.utils import find_files, load_file
 
 SCHEMA_TAG = "jsonschema"
@@ -23,7 +24,6 @@ def __init__(self, config):
         self.config = config
 
         # Find all instance files
-        # TODO need to load file extensions from the config
         instance_files = find_files(
             file_extensions=config.data_file_extensions,
             search_directories=config.data_file_search_directories,
@@ -97,7 +97,24 @@ def top_level_properties(self):
         """
         if not self._top_level_properties:
             content = self._get_content()
-            self._top_level_properties = set(content.keys())
+            # TODO: Investigate and see if we should be checking this on initialization if the file doesn't exists or is empty.
+            if not content:
+                return self._top_level_properties
+
+            if isinstance(content, CommentedMap) or hasattr(content, "keys"):
+                self._top_level_properties = set(content.keys())
+            elif isinstance(content, str):
+                self._top_level_properties = set([content])
+            elif isinstance(content, list):
+                properties = set()
+                for m in content:
+                    if isinstance(m, dict) or hasattr(m, "keys"):
+                        properties.update(m.keys())
+                    else:
+                        properties.add(m)
+                self._top_level_properties = properties
+            else:
+                self._top_level_properties = set(content)
 
         return self._top_level_properties
 

schema_enforcer/schemas/manager.py

@@ -7,6 +7,9 @@
 from termcolor import colored
 from rich.console import Console
 from rich.table import Table
+from typing import List, Optional, Type
+
+from pydantic import BaseModel
 
 from schema_enforcer.utils import load_file, find_file, find_files, dump_data_to_yaml
 from schema_enforcer.validation import ValidationResult, RESULT_PASS, RESULT_FAIL
@@ -45,7 +48,7 @@ def __init__(self, config):
             self.schemas[schema.get_id()] = schema
 
         # Load validators
-        validators = load_validators(config.validator_directory)
+        validators = load_validators(config.validator_directory, config.pydantic_validators)
         self.schemas.update(validators)
 
     def create_schema_from_file(self, root, filename):
@@ -94,7 +97,12 @@ def print_schemas_list(self):
         table.add_column("Location")
         table.add_column("Filename")
         for schema_id, schema in self.iter_schemas():
-            table.add_row(schema_id, schema.schematype, schema.root.replace(current_dir, "."), schema.filename)
+            table.add_row(
+                schema_id,
+                schema.schematype,
+                schema.root.replace(current_dir, "."),
+                schema.filename,
+            )
         console.print(table)
 
     def dump_schema(self, schema_id=None):
@@ -336,3 +344,10 @@ def _ensure_results_invalid(results, data_file):
         if "PASS" in results_pass_or_fail:
             error(f"{data_file} is schema valid, but should be schema invalid as it defines an invalid test")
             sys.exit(1)
+
+
+class PydanticManager(BaseModel):
+    """Class for managing Pydantic models and adding them to the SchemaManager."""
+
+    prefix: Optional[str] = ""
+    models: List[Type[BaseModel]]