Add data validation support using Pydantic models.

Author: fabiocaccamo

README.md

@@ -65,6 +65,7 @@ Here the hierarchy of possible installation targets available when running `pip
         - `[yaml]`
     - `[parse]`
     - `[s3]`
+    - `[validate]`
 
 ## Usage
 
@@ -279,6 +280,36 @@ Here are the details of the supported formats, operations and extra options docs
 | `xml`          | :white_check_mark: | :white_check_mark: | [xmltodict](https://github.com/martinblech/xmltodict)                                 |
 | `yaml`         | :white_check_mark: | :white_check_mark: | [PyYAML](https://pyyaml.org/wiki/PyYAMLDocumentation)                                 |
 
+#### Data validation
+
+`benedict` supports data validation using `Pydantic` models.
+
+This feature **requires** the `validate` extra to be installed:
+
+```bash
+pip install "python-benedict[validate]"
+```
+
+You can validate data in different ways:
+
+##### Using the `validate` method directly
+```python
+d = benedict(my_data)
+d.validate(schema=MySchema)
+```
+
+##### Using the `schema` parameter during initialization
+```python
+d = benedict(my_data, schema=MySchema)
+```
+
+##### Using the `schema` parameter with any `from_{format}` method
+```python
+d = benedict.from_json(my_data, schema=MySchema)
+```
+
+If validation fails, a `ValidationError` will be raised with details about what went wrong.
+
 ### API
 
 -   **Utility methods**
@@ -333,6 +364,7 @@ Here are the details of the supported formats, operations and extra options docs
     -   [`to_toml`](#to_toml)
     -   [`to_xml`](#to_xml)
     -   [`to_yaml`](#to_yaml)
+    -   [`validate`](#validate)
 
 -   **Parse methods**
 
@@ -815,6 +847,14 @@ s = d.to_xml(**kwargs)
 s = d.to_yaml(**kwargs)
 ```
 
+#### `validate`
+
+```python
+# Validate the dict and update it using a Pydantic schema.
+# A ValidationError is raised in case of failure.
+d.validate(schema=MySchema)
+```
+
 ### Parse methods
 
 These methods are wrappers of the `get` method, they parse data trying to return it in the expected type.

benedict/dicts/io/io_dict.py

@@ -3,7 +3,8 @@
 from benedict.dicts.base import BaseDict
 from benedict.dicts.io import io_util
 from benedict.exceptions import ExtrasRequireModuleNotFoundError
-from benedict.utils import type_util
+from benedict.utils import pydantic_util, type_util
+from benedict.utils.pydantic_util import PydanticModel
 
 
 class IODict(BaseDict):
@@ -20,18 +21,24 @@ def __init__(self, *args, **kwargs):
                 d = IODict._decode_init(arg, **kwargs)
                 super().__init__(d)
                 return
+
+        schema = kwargs.pop("schema", None)
         super().__init__(*args, **kwargs)
+        if schema:
+            self.validate(schema=schema)
 
     @staticmethod
     def _decode_init(s, **kwargs):
         autodetected_format = io_util.autodetect_format(s)
         default_format = autodetected_format or "json"
         format = kwargs.pop("format", default_format).lower()
         # decode data-string and initialize with dict data.
-        return IODict._decode(s, format, **kwargs)
+        data = IODict._decode(s, format, **kwargs)
+        return data
 
     @staticmethod
     def _decode(s, format, **kwargs):
+        schema = kwargs.pop("schema", None)
         data = None
         try:
             data = io_util.decode(s, format, **kwargs)
@@ -45,12 +52,15 @@ def _decode(s, format, **kwargs):
             ) from None
         # if possible return data as dict, otherwise raise exception
         if type_util.is_dict(data):
-            return data
+            pass
         elif type_util.is_list(data):
             # force list to dict
-            return {"values": data}
+            data = {"values": data}
         else:
             raise ValueError(f"Invalid data type: {type(data)}, expected dict or list.")
+        if schema:
+            data = pydantic_util.validate_data(data, schema=schema)
+        return data
 
     @staticmethod
     def _encode(d, format, **kwargs):
@@ -325,3 +335,14 @@ def to_yaml(self, **kwargs):
         A ValueError is raised in case of failure.
         """
         return self._encode(self.dict(), "yaml", **kwargs)
+
+    def validate(self, *, schema: PydanticModel):
+        """
+        Validate the dict and update it using a Pydantic schema.
+
+        Args:
+            schema: Pydantic model class for validation
+        """
+        data = pydantic_util.validate_data(self, schema=schema)
+        self.clear()
+        self.update(data)

benedict/extras.py

@@ -5,6 +5,7 @@
     "require_parse",
     "require_s3",
     "require_toml",
+    "require_validate",
     "require_xls",
     "require_xml",
     "require_yaml",
@@ -32,6 +33,10 @@ def require_toml(*, installed):
     _require_optional_dependencies(target="toml", installed=installed)
 
 
+def require_validate(*, installed):
+    _require_optional_dependencies(target="validate", installed=installed)
+
+
 def require_xls(*, installed):
     _require_optional_dependencies(target="xls", installed=installed)
 

benedict/utils/pydantic_util.py

@@ -0,0 +1,50 @@
+from typing import Any
+
+from benedict.extras import require_validate
+
+try:
+    from pydantic.v2 import BaseModel
+    from pydantic.v2.json import pydantic_encoder
+
+    pydantic_installed = True
+except ImportError:
+    pydantic_installed = False
+    BaseModel = None
+    pydantic_encoder = None
+
+PydanticModel = type["BaseModel"]
+
+
+def is_pydantic_model(obj: Any) -> bool:
+    """
+    Check if an object is a Pydantic model.
+    """
+    return pydantic_installed and isinstance(obj, BaseModel)
+
+
+def is_pydantic_model_class(obj: Any) -> bool:
+    """
+    Check if an object is a Pydantic model class.
+    """
+    return (
+        pydantic_installed
+        and BaseModel is not None
+        and isinstance(obj, type)
+        and issubclass(obj, BaseModel)
+    )
+
+
+def validate_data(data: Any, *, schema: PydanticModel | None = None) -> Any:
+    """
+    Validate data against a Pydantic schema if provided.
+    """
+    if schema is None:
+        return data
+
+    require_validate(installed=pydantic_installed)
+
+    if not is_pydantic_model_class(schema):
+        raise ValueError("Invalid schema. Schema must be a Pydantic model class.")
+
+    validated = schema.model_validate(data)
+    return validated.model_dump()

pyproject.toml

@@ -115,7 +115,7 @@ Twitter = "https://twitter.com/fabiocaccamo"
 
 [project.optional-dependencies]
 all = [
-    "python-benedict[io,parse,s3]",
+    "python-benedict[io,parse,s3,validate]",
 ]
 html = [
     "beautifulsoup4 >= 4.12.0, < 5.0.0",
@@ -136,6 +136,9 @@ s3 = [
 toml = [
     "toml >= 0.10.2, < 1.0.0",
 ]
+validate = [
+    "pydantic >= 1.10.0",
+]
 xls = [
     "openpyxl >= 3.0.0, < 4.0.0",
     "xlrd >= 2.0.0, < 3.0.0",

tests/dicts/io/test_io_dict_schema.py

@@ -0,0 +1,86 @@
+import json
+import unittest
+
+from benedict import benedict
+
+try:
+    from pydantic.v2 import BaseModel
+    from pydantic.v2.errors import ValidationError
+
+    pydantic_installed = True
+except ImportError:
+    pydantic_installed = False
+    BaseModel = None
+    ValidationError = None
+
+
+@unittest.skipIf(not pydantic_installed, "pydantic not installed")
+class TestIODictSchema(unittest.TestCase):
+    def setUp(self):
+        class User(BaseModel):
+            name: str
+            age: int
+            email: str
+
+        class UserList(BaseModel):
+            users: list[User]
+
+        class UserOptional(BaseModel):
+            name: str
+            age: int | None = None
+            email: str | None = None
+
+        self.User = User
+        self.UserList = UserList
+        self.UserOptional = UserOptional
+        self.valid_data = {
+            "name": "John",
+            "age": 30,
+            "email": "john@example.com",
+        }
+        self.invalid_data = {
+            "name": "John",
+            "age": "not_an_int",
+            "email": "john@example.com",
+        }
+        self.minimal_data = {"name": "John"}
+
+    def test_constructor_with_schema(self):
+        d = benedict(self.valid_data, schema=self.User)
+        self.assertEqual(d["name"], "John")
+        self.assertEqual(d["age"], 30)
+        self.assertEqual(d["email"], "john@example.com")
+
+        with self.assertRaises(ValidationError):
+            benedict(self.invalid_data, schema=self.User)
+
+    def test_constructor_with_schema_and_optional_fields(self):
+        d = benedict(self.minimal_data, schema=self.UserOptional)
+        self.assertEqual(d["name"], "John")
+        self.assertIsNone(d.get("age"))
+        self.assertIsNone(d.get("email"))
+
+    def test_constructor_with_invalid_schema(self):
+        class InvalidSchema:
+            pass
+
+        with self.assertRaises(ValueError):
+            benedict(self.valid_data, schema=InvalidSchema)
+
+        with self.assertRaises(ValueError):
+            benedict(self.valid_data, schema="not_a_schema")
+
+    def test_from_json_with_schema_and_valid_data(self):
+        json_data = json.dumps(self.valid_data)
+        d = benedict.from_json(json_data, schema=self.User)
+        self.assertEqual(d["name"], "John")
+        self.assertEqual(d["age"], 30)
+
+    def test_from_json_with_schema_and_invalid_data(self):
+        json_data = json.dumps(self.invalid_data)
+        with self.assertRaises(ValidationError):
+            benedict.from_json(json_data, schema=self.User)
+
+
+if __name__ == "__main__":
+    unittest.main()