Add experimental model_validate function for Pydantic V2 compatibil…

…ity (PrefectHQ#12370)

Co-authored-by: Serina Grill <[email protected]>

src/prefect/_internal/pydantic/__init__.py

@@ -10,4 +10,4 @@
 
 HAS_PYDANTIC_V2 = PYDANTIC_VERSION.startswith("2.")
 
-from ._compat import model_dump, IncEx, model_json_schema
+from ._compat import model_dump, model_json_schema, model_validate, IncEx

src/prefect/_internal/pydantic/_compat.py

@@ -178,3 +178,36 @@ def model_json_schema(
         by_alias=by_alias,
         ref_template=ref_template,
     )
+
+
+def model_validate(
+    model: Type[BaseModel],
+    obj: Any,
+    *,
+    strict: bool = False,
+    from_attributes: bool = False,
+    context: Optional[Dict[str, Any]] = None,
+) -> Union[BaseModel, Dict[str, Any]]:
+    """Validate a pydantic model instance.
+
+    Args:
+        obj: The object to validate.
+        strict: Whether to enforce types strictly.
+        from_attributes: Whether to extract data from object attributes.
+        context: Additional context to pass to the validator.
+
+    Raises:
+        ValidationError: If the object could not be validated.
+
+    Returns:
+        The validated model instance.
+    """
+    if is_pydantic_v2_compatible(fn_name="model_validate"):
+        return model.model_validate(
+            obj=obj,
+            strict=strict,
+            from_attributes=from_attributes,
+            context=context,
+        )
+
+    return model.parse_obj(obj)

tests/_internal/pydantic/test_model_validate.py

@@ -0,0 +1,69 @@
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from prefect._internal.pydantic._compat import HAS_PYDANTIC_V2, model_validate
+from prefect.settings import (
+    PREFECT_EXPERIMENTAL_ENABLE_PYDANTIC_V2_INTERNALS,
+    temporary_settings,
+)
+
+
+@pytest.fixture(autouse=True)
+def enable_v2_internals():
+    with temporary_settings({PREFECT_EXPERIMENTAL_ENABLE_PYDANTIC_V2_INTERNALS: True}):
+        yield
+
+
+class Model(BaseModel):
+    a: int
+    b: str
+
+
+def test_model_validate(caplog):
+    model_instance = model_validate(Model, {"a": 1, "b": "test"})
+
+    assert model_instance.a == 1
+
+    assert model_instance.b == "test"
+
+    if HAS_PYDANTIC_V2:
+        assert (
+            "Using Pydantic v2 compatibility layer for `model_validate`" in caplog.text
+        )
+
+    else:
+        assert "Pydantic v2 is not installed." in caplog.text
+
+
+def test_model_validate_with_flag_disabled(caplog):
+    with temporary_settings({PREFECT_EXPERIMENTAL_ENABLE_PYDANTIC_V2_INTERNALS: False}):
+        if HAS_PYDANTIC_V2:
+            from pydantic.warnings import PydanticDeprecatedSince20
+
+            with pytest.warns(PydanticDeprecatedSince20):
+                model_instance = model_validate(Model, {"a": 1, "b": "test"})
+        else:
+            model_instance = model_validate(Model, {"a": 1, "b": "test"})
+
+    assert model_instance.a == 1
+
+    assert model_instance.b == "test"
+
+    if HAS_PYDANTIC_V2:
+        assert "Pydantic v2 compatibility layer is disabled" in caplog.text
+    else:
+        assert "Pydantic v2 is not installed." in caplog.text
+
+
+def test_model_validate_with_invalid_model(caplog):
+    try:
+        model_validate(Model, {"a": "not an int", "b": "test"})
+    except ValidationError as e:
+        errors = e.errors()
+
+        assert len(errors) == 1
+
+        error = errors[0]
+
+        assert error["loc"] == ("a",)
+        assert "valid integer" in error["msg"]