chore!: Update model parsing to use Pydantic V2 instead of Pydantic V1

Signed-off-by: Mark Wiebe <[email protected]>

Author: mwiebe

pyproject.toml

@@ -30,7 +30,7 @@ classifiers = [
 ]
 dependencies = [
   "pyyaml == 6.0.*",
-  "pydantic >= 1.10.17",
+  "pydantic >= 2.10, < 3",
 ]
 
 [project.urls]
@@ -99,7 +99,7 @@ mypy_path = "src"
 
 # See: https://docs.pydantic.dev/mypy_plugin/
 #  - Helps mypy understand pydantic typing.
-plugins = "pydantic.v1.mypy"
+plugins = "pydantic.mypy"
 
 [tool.ruff]
 line-length = 100
@@ -111,15 +111,6 @@ ignore = [
   "E731",
 ]
 
-[tool.ruff.lint.pep8-naming]
-classmethod-decorators = [
-  "classmethod",
-  # pydantic decorators are classmethod decorators
-  # suppress N805 errors on classes decorated with them
-  "pydantic.validator",
-  "pydantic.root_validator",
-]
-
 [tool.ruff.lint.isort]
 known-first-party = [
   "openjd",

src/openjd/model/_convert_pydantic_error.py

@@ -1,37 +1,33 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
-from typing import TypedDict, Union, Type
-from pydantic.v1 import BaseModel
+from typing import Type, Union
+from pydantic import BaseModel
+from pydantic_core import ErrorDetails
 from inspect import getmodule
 
-# Calling pydantic's ValidationError.errors() returns a list[ErrorDict], but
-# pydantic doesn't export the ErrorDict type publicly. So, we create it here for
-# type checking.
-# Note that we ignore the 'ctx' key since we don't use it.
-# See: https://github.com/pydantic/pydantic/blob/d9c2af3a701ca982945a590de1a1da98b3fb4003/pydantic/error_wrappers.py#L50
-Loc = tuple[Union[int, str], ...]
 
-
-class ErrorDict(TypedDict):
-    loc: Loc
-    msg: str
-    type: str
-
-
-def pydantic_validationerrors_to_str(root_model: Type[BaseModel], errors: list[ErrorDict]) -> str:
+def pydantic_validationerrors_to_str(
+    root_model: Type[BaseModel], errors: list[ErrorDetails]
+) -> str:
     """This is our own custom stringification of the Pydantic ValidationError to use
     in place of str(<ValidationError>). Pydantic's default stringification too verbose for
     our purpose, and contains information that we don't want.
     """
     results = list[str]()
-    for error in errors:
-        results.append(_error_dict_to_str(root_model, error))
+    for error_details in errors:
+        results.append(_error_dict_to_str(root_model, error_details))
     return f"{len(errors)} validation errors for {root_model.__name__}\n" + "\n".join(results)
 
 
-def _error_dict_to_str(root_model: Type[BaseModel], error: ErrorDict) -> str:
-    loc = error["loc"]
-    msg = error["msg"]
+def _error_dict_to_str(root_model: Type[BaseModel], error_details: ErrorDetails) -> str:
+    error_type = error_details["type"]
+    loc = error_details["loc"]
+    # Skip the "Value error," prefix by getting the exception message directly.
+    # This preserves the message formatting created when Pydantic V1 was in use.
+    if error_type == "value_error":
+        msg = str(error_details["ctx"]["error"])
+    else:
+        msg = error_details["msg"]
 
     # When a model's root_validator raises an error other than a ValidationError
     # (i.e. raises something like a ValueError or a TypeError) then pydantic
@@ -54,7 +50,7 @@ def _error_dict_to_str(root_model: Type[BaseModel], error: ErrorDict) -> str:
     return f"{_loc_to_str(root_model, loc)}:\n\t{msg}"
 
 
-def _loc_to_str(root_model: Type[BaseModel], loc: Loc) -> str:
+def _loc_to_str(root_model: Type[BaseModel], loc: tuple[Union[int, str], ...]) -> str:
     model_module = getmodule(root_model)
 
     # If a nested error is from a root validator, then just report the error as being

src/openjd/model/_create_job.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from typing import Optional, cast
 
-from pydantic.v1 import ValidationError
+from pydantic import ValidationError
 
 from ._errors import CompatibilityError, DecodeValidationError
 from ._symbol_table import SymbolTable
@@ -22,7 +22,7 @@
     SpecificationRevision,
     TemplateSpecificationVersion,
 )
-from ._convert_pydantic_error import pydantic_validationerrors_to_str, ErrorDict
+from ._convert_pydantic_error import pydantic_validationerrors_to_str
 
 __all__ = ("preprocess_job_parameters",)
 
@@ -330,9 +330,7 @@ def create_job(
         job = instantiate_model(job_template, symtab)
     except ValidationError as exc:
         raise DecodeValidationError(
-            pydantic_validationerrors_to_str(
-                job_template.__class__, cast(list[ErrorDict], exc.errors())
-            )
+            pydantic_validationerrors_to_str(job_template.__class__, exc.errors())
         )
 
     return cast(Job, job)

src/openjd/model/_format_strings/_dyn_constrained_str.py

@@ -1,14 +1,11 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
-import re
-from typing import TYPE_CHECKING, Any, Callable, Optional, Pattern, Union
-
-from pydantic.v1.errors import AnyStrMaxLengthError, AnyStrMinLengthError, StrRegexError
-from pydantic.v1.utils import update_not_none
-from pydantic.v1.validators import strict_str_validator
+from typing import Any, Callable, Optional, Pattern, Union
 
-if TYPE_CHECKING:
-    from pydantic.v1.typing import CallableGenerator
+from pydantic import GetCoreSchemaHandler, GetJsonSchemaHandler
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import core_schema
+import re
 
 
 class DynamicConstrainedStr(str):
@@ -33,37 +30,43 @@ def _get_max_length(cls) -> Optional[int]:
         return cls._max_length
 
     @classmethod
-    def __modify_schema__(cls, field_schema: dict[str, Any]) -> None:
-        update_not_none(
-            field_schema,
-            minLength=cls._min_length,
-            maxLength=cls._get_max_length(),
-        )
-
-    @classmethod
-    def __get_validators__(cls) -> "CallableGenerator":
-        yield strict_str_validator  # Always strict string.
-        yield cls._validate_min_length
-        yield cls._validate_max_length
-        yield cls._validate_regex
+    def _validate(cls, value: str) -> Any:
+        if not isinstance(value, str):
+            raise ValueError("String required")
 
-    @classmethod
-    def _validate_min_length(cls, value: str) -> str:
         if cls._min_length is not None and len(value) < cls._min_length:
-            raise AnyStrMinLengthError(limit_value=cls._min_length)
-        return value
-
-    @classmethod
-    def _validate_max_length(cls, value: str) -> str:
+            raise ValueError(f"String must be at least {cls._min_length} characters long")
         max_length = cls._get_max_length()
+
         if max_length is not None and len(value) > max_length:
-            raise AnyStrMaxLengthError(limit_value=max_length)
-        return value
+            raise ValueError(f"String must be at most {max_length} characters long")
 
-    @classmethod
-    def _validate_regex(cls, value: str) -> str:
         if cls._regex is not None:
             if not re.match(cls._regex, value):
                 pattern: str = cls._regex if isinstance(cls._regex, str) else cls._regex.pattern
-                raise StrRegexError(pattern=pattern)
-        return value
+                raise ValueError(f"String does not match the required pattern: {pattern}")
+
+        return cls(value)
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source_type: type[Any], handler: GetCoreSchemaHandler
+    ) -> core_schema.CoreSchema:
+        return core_schema.no_info_plain_validator_function(cls._validate)
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
+    ) -> JsonSchemaValue:
+        json_schema: dict[str, Any] = {"type": "string"}
+        if cls._min_length is not None:
+            json_schema["minLength"] = cls._min_length
+        max_length = cls._get_max_length()
+        if max_length is not None:
+            json_schema["maxLength"] = max_length
+        if cls._regex is not None:
+            json_schema["pattern"] = (
+                cls._regex if isinstance(cls._regex, str) else cls._regex.pattern
+            )
+
+        return json_schema

src/openjd/model/_format_strings/_format_string.py

@@ -2,16 +2,13 @@
 
 from dataclasses import dataclass
 from numbers import Real
-from typing import TYPE_CHECKING, Optional, Union
+from typing import Optional, Union
 
 from .._errors import ExpressionError, TokenError
 from .._symbol_table import SymbolTable
 from ._dyn_constrained_str import DynamicConstrainedStr
 from ._expression import InterpolationExpression
 
-if TYPE_CHECKING:
-    from pydantic.v1.typing import CallableGenerator
-
 
 @dataclass
 class ExpressionInfo:
@@ -21,8 +18,9 @@ class ExpressionInfo:
     resolved_value: Optional[Union[Real, str]] = None
 
 
-class FormatStringError(Exception):
+class FormatStringError(ValueError):
     def __init__(self, *, string: str, start: int, end: int, expr: str = "", details: str = ""):
+        self.input = string
         expression = f"Expression: {expr}. " if expr else ""
         reason = f"Reason: {details}." if details else ""
         msg = (
@@ -202,24 +200,3 @@ def _preprocess(self) -> list[Union[str, ExpressionInfo]]:
             result_list.append(expression_info)
 
         return result_list
-
-    # Pydantic datamodel interfaces
-    # ================================
-    # Reference: https://pydantic-docs.helpmanual.io/usage/types/#custom-data-types
-
-    @classmethod
-    def __get_validators__(cls) -> "CallableGenerator":
-        for validator in super().__get_validators__():
-            yield validator
-        yield cls._validate
-
-    @classmethod
-    def _validate(cls, value: str) -> "FormatString":
-        # Reference: https://pydantic-docs.helpmanual.io/usage/validators/
-        # Class constructor will raise validation errors on the value contents.
-        try:
-            return cls(value)
-        except FormatStringError as e:
-            # Pydantic validators must return a ValueError or AssertionError
-            # Convert the FormatStringError into a ValueError
-            raise ValueError(str(e))