[Train] Replace Checkpoint Manager pydantic v2 APIs with v1 APIs (#57147)

Replace `CheckpointManager`'s usage of pydantic v2 APIs with v1 APIs instead.

---------

Signed-off-by: JasonLi1909 <jasli1909@gmail.com>
Signed-off-by: Jason Li <57246540+JasonLi1909@users.noreply.github.com>
Co-authored-by: Justin Yu <justinvyu@anyscale.com>

Author: JasonLi1909

python/ray/train/__init__.py

@@ -28,6 +28,13 @@
 from ray.train.v2._internal.constants import is_v2_enabled
 
 if is_v2_enabled():
+    try:
+        import pydantic  # noqa: F401
+    except (ImportError, ModuleNotFoundError) as exc:
+        raise ImportError(
+            "`ray.train.v2` requires the pydantic package, which is missing. "
+            "Run the following command to fix this: `pip install pydantic`"
+        ) from exc
     from ray.train.v2.api.callback import UserCallback  # noqa: F811
     from ray.train.v2.api.config import (  # noqa: F811
         FailureConfig,

python/ray/train/v2/_internal/execution/checkpoint/checkpoint_manager.py

@@ -1,7 +1,9 @@
 import asyncio
+import json
 import logging
 from typing import Any, Dict, List, Optional
 
+from ray._common.pydantic_compat import BaseModel
 from ray.air.config import CheckpointConfig
 from ray.train._checkpoint import Checkpoint
 from ray.train._internal.checkpoint_manager import (
@@ -20,16 +22,6 @@
 from ray.train.v2._internal.execution.worker_group import Worker
 from ray.train.v2.api.reported_checkpoint import ReportedCheckpoint
 
-try:
-    from pydantic import BaseModel
-    from pydantic_core import from_json
-except (ImportError, ModuleNotFoundError) as exc:
-    raise ImportError(
-        "`ray.train.v2` requires the pydantic package, which is missing. "
-        "Run the following command to fix this: `pip install pydantic`"
-    ) from exc
-
-
 logger = logging.getLogger(__name__)
 
 
@@ -235,14 +227,13 @@ def _save_state(self) -> str:
             checkpoint_results=checkpoint_results,
             latest_checkpoint_result=latest_checkpoint_result,
         )
-        return manager_snapshot.model_dump_json()
+        return manager_snapshot.json()
 
     def _load_state(self, json_state: str):
         """Load the checkpoint manager state from a JSON str."""
         try:
-            manager_snapshot = _CheckpointManagerState.model_validate(
-                from_json(json_state)
-            )
+            json_dict = json.loads(json_state)
+            manager_snapshot = _CheckpointManagerState.parse_obj(json_dict)
         except Exception as e:
             raise CheckpointManagerInitializationError(repr(e)) from e
         self._assert_checkpoints_exist()