fix: handle pydantic generic models in JSON serializer

When using parameterized Pydantic generic models like `APIResult[str]`,
the JSON serializer was storing class names with brackets (e.g.,
`module.APIResult[str]`) which cannot be imported by Python's import
system, causing deserialization to fail.

This fix extracts the origin class from Pydantic's generic metadata,
ensuring the serialized class name is importable (e.g., `module.APIResult`).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: zzstoatzz

src/prefect/serializers.py

@@ -41,6 +41,21 @@
 _TYPE_ADAPTER_CACHE: dict[str, TypeAdapter[Any]] = {}
 
 
+def _get_importable_class(cls: type) -> type:
+    """
+    Get an importable class from a potentially parameterized generic.
+
+    For Pydantic generic models like `APIResult[str]`, the class name includes
+    type parameters (e.g., `APIResult[str]`) which cannot be imported. This
+    function extracts the origin class (e.g., `APIResult`) which can be imported.
+    """
+    if hasattr(cls, "__pydantic_generic_metadata__"):
+        origin = cls.__pydantic_generic_metadata__.get("origin")
+        if origin is not None:
+            return origin
+    return cls
+
+
 def prefect_json_object_encoder(obj: Any) -> Any:
     """
     `JSONEncoder.default` for encoding objects into JSON with extended type support.
@@ -58,8 +73,9 @@ def prefect_json_object_encoder(obj: Any) -> Any:
             ),
         }
     else:
+        importable_class = _get_importable_class(obj.__class__)
         return {
-            "__class__": to_qualified_name(obj.__class__),
+            "__class__": to_qualified_name(importable_class),
             "data": custom_pydantic_encoder({}, obj),
         }
 

tests/test_serializers.py

@@ -3,7 +3,7 @@
 import json
 import uuid
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Generic, TypeVar
 from unittest.mock import MagicMock
 
 import pytest
@@ -30,6 +30,16 @@ class MyModel(BaseModel):
     y: uuid.UUID
 
 
+T = TypeVar("T")
+
+
+class GenericResult(BaseModel, Generic[T]):
+    """Generic model for testing JSON serialization of parameterized types."""
+
+    data: T | None = None
+    message: str = ""
+
+
 @dataclass
 class MyDataclass:
     x: int
@@ -387,6 +397,32 @@ def test_does_not_allow_default_collision(self):
         with pytest.raises(ValidationError):
             JSONSerializer(dumps_kwargs={"default": "foo"})
 
+    def test_pydantic_generic_model_roundtrip(self):
+        """Test that Pydantic generic models with type parameters can be serialized.
+
+        Regression test for: https://github.com/PrefectHQ/prefect/issues/XXXX
+
+        When using parameterized generics like `APIResult[str]`, the class name
+        includes brackets which cannot be imported. The serializer should extract
+        the origin class for proper roundtrip serialization.
+        """
+        serializer = JSONSerializer()
+
+        # Test with concrete type parameter
+        result = GenericResult[str](data="hello", message="success")
+        serialized = serializer.dumps(result)
+
+        # Verify the serialized class name doesn't include type parameters
+        decoded = json.loads(serialized)
+        assert "[" not in decoded["__class__"], (
+            f"Class name should not contain brackets: {decoded['__class__']}"
+        )
+
+        # Verify roundtrip works
+        loaded = serializer.loads(serialized)
+        assert loaded.data == "hello"
+        assert loaded.message == "success"
+
 
 class TestCompressedSerializer:
     @pytest.mark.parametrize("data", SERIALIZER_TEST_CASES)