fix(partial): prevent infinite recursion with self-referential models (#1997)

- Add _processing_models set to track recursive models
- Prevent infinite recursion with self-referential models (e.g., TreeNode with children: List["TreeNode"])
- Add tests for recursive model handling

Co-Authored-By: Thomas Dybdahl Ahle <thomas@ahle.dk>
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: 3 people

instructor/dsl/partial.py

@@ -41,6 +41,10 @@
 else:
     UNION_ORIGINS = (Union,)
 
+# Track models currently being processed to prevent infinite recursion
+# with self-referential models (e.g., TreeNode with children: List["TreeNode"])
+_processing_models: set[type] = set()
+
 
 class MakeFieldsOptional:
     pass
@@ -102,11 +106,18 @@ def _process_generic_arg(
         return arg_origin[modified_nested_args]
     else:
         if isinstance(arg, type) and issubclass(arg, BaseModel):
-            return (
-                Partial[arg, MakeFieldsOptional]  # type: ignore[valid-type]
-                if make_fields_optional
-                else Partial[arg]
-            )
+            # Prevent infinite recursion for self-referential models
+            if arg in _processing_models:
+                return arg  # Already processing this model, return unwrapped
+            _processing_models.add(arg)
+            try:
+                return (
+                    Partial[arg, MakeFieldsOptional]  # type: ignore[valid-type]
+                    if make_fields_optional
+                    else Partial[arg]
+                )
+            finally:
+                _processing_models.discard(arg)
         else:
             return arg
 
@@ -946,7 +957,17 @@ def _wrap_models(field: FieldInfo) -> tuple[object, FieldInfo]:
             # If the field is a BaseModel, then recursively convert it's
             # attributes to optionals.
             elif isinstance(annotation, type) and issubclass(annotation, BaseModel):
-                tmp_field.annotation = Partial[annotation]
+                # Prevent infinite recursion for self-referential models
+                if annotation in _processing_models:
+                    tmp_field.annotation = (
+                        annotation  # Already processing, keep unwrapped
+                    )
+                else:
+                    _processing_models.add(annotation)
+                    try:
+                        tmp_field.annotation = Partial[annotation]
+                    finally:
+                        _processing_models.discard(annotation)
             return tmp_field.annotation, tmp_field
 
         model_name = (

tests/dsl/test_partial.py

@@ -101,7 +101,9 @@ def test_partial_with_whitespace():
     partial = Partial[SamplePartial]
 
     # Get the actual models from chunks - must provide complete data for final validation
-    models = list(partial.model_from_chunks(["\n", "\t", " ", '{"a": 42, "b": {"b": 1}}']))
+    models = list(
+        partial.model_from_chunks(["\n", "\t", " ", '{"a": 42, "b": {"b": 1}}'])
+    )
 
     # Print actual values for debugging
     print(f"Number of models: {len(models)}")
@@ -882,3 +884,241 @@ async def async_chunks():
                 pass
 
         assert "age" in str(exc_info.value)
+
+
+class TestRecursiveModels:
+    """Test that Partial handles self-referential models without infinite recursion."""
+
+    def test_basic_recursive_model(self):
+        """Partial should work with basic recursive models."""
+
+        class TreeNode(BaseModel):
+            value: str
+            children: Optional[list["TreeNode"]] = None
+
+        TreeNode.model_rebuild()
+
+        # Should not raise RecursionError
+        PartialTreeNode = Partial[TreeNode]
+        TruePartial = PartialTreeNode.get_partial_model()
+
+        # Can validate partial data
+        result = TruePartial.model_validate({"value": "root"})
+        assert result.value == "root"
+        assert result.children is None
+
+    def test_nested_recursive_model(self):
+        """Partial should work with nested children."""
+
+        class TreeNode(BaseModel):
+            value: str
+            children: Optional[list["TreeNode"]] = None
+
+        TreeNode.model_rebuild()
+
+        PartialTreeNode = Partial[TreeNode]
+        TruePartial = PartialTreeNode.get_partial_model()
+
+        # Validate with nested structure
+        data = {
+            "value": "root",
+            "children": [
+                {"value": "child1"},
+                {"value": "child2", "children": [{"value": "grandchild"}]},
+            ],
+        }
+        result = TruePartial.model_validate(data)
+        assert result.value == "root"
+        assert len(result.children) == 2
+        assert result.children[0].value == "child1"
+        assert result.children[1].children[0].value == "grandchild"
+
+    def test_mutually_recursive_models(self):
+        """Partial should handle mutually recursive models."""
+
+        class Person(BaseModel):
+            name: str
+            employer: Optional["Company"] = None
+
+        class Company(BaseModel):
+            name: str
+            employees: Optional[list[Person]] = None
+
+        Person.model_rebuild()
+        Company.model_rebuild()
+
+        # Both should work without RecursionError
+        PartialPerson = Partial[Person]
+        PartialCompany = Partial[Company]
+
+        assert PartialPerson is not None
+        assert PartialCompany is not None
+
+        # Validate partial data
+        person_partial = PartialPerson.get_partial_model()
+        result = person_partial.model_validate({"name": "Alice"})
+        assert result.name == "Alice"
+
+    def test_direct_self_reference(self):
+        """Partial should handle direct self-reference (linked list style)."""
+
+        class LinkedNode(BaseModel):
+            value: int
+            next: Optional["LinkedNode"] = None
+
+        LinkedNode.model_rebuild()
+
+        # Should not raise RecursionError
+        PartialLinked = Partial[LinkedNode]
+        TruePartial = PartialLinked.get_partial_model()
+
+        # Validate chain
+        data = {"value": 1, "next": {"value": 2, "next": {"value": 3}}}
+        result = TruePartial.model_validate(data)
+        assert result.value == 1
+        assert result.next.value == 2
+        assert result.next.next.value == 3
+
+    def test_complex_recursive_with_validators(self):
+        """Complex recursive model with validators, multiple self-refs, and nested types."""
+        from typing import Literal
+        from pydantic import model_validator, field_validator
+        from enum import Enum
+
+        class NodeType(Enum):
+            FOLDER = "folder"
+            FILE = "file"
+            SYMLINK = "symlink"
+
+        class Permission(BaseModel):
+            user: str
+            level: Literal["read", "write", "admin"]
+
+        class FileSystemNode(BaseModel):
+            name: str
+            node_type: NodeType
+            size_bytes: Optional[int] = None
+            children: Optional[list["FileSystemNode"]] = None
+            parent: Optional["FileSystemNode"] = None
+            symlink_target: Optional["FileSystemNode"] = None
+            permissions: Optional[list[Permission]] = None
+            metadata: Optional[dict[str, str]] = None
+
+            @field_validator("name")
+            @classmethod
+            def validate_name(cls, v):
+                if v and "/" in v:
+                    raise ValueError("Name cannot contain /")
+                return v
+
+            @model_validator(mode="after")
+            def validate_node_consistency(self):
+                # Folders must have no size, files must have size
+                if self.node_type == NodeType.FOLDER and self.size_bytes is not None:
+                    raise ValueError("Folders cannot have size_bytes")
+                if self.node_type == NodeType.FILE and self.children:
+                    raise ValueError("Files cannot have children")
+                if self.node_type == NodeType.SYMLINK and not self.symlink_target:
+                    raise ValueError("Symlinks must have a target")
+                return self
+
+        FileSystemNode.model_rebuild()
+
+        # Should not raise RecursionError
+        PartialFS = Partial[FileSystemNode]
+        TruePartial = PartialFS.get_partial_model()
+
+        # Complex nested structure
+        data = {
+            "name": "root",
+            "node_type": "folder",
+            "permissions": [{"user": "admin", "level": "admin"}],
+            "metadata": {"created": "2024-01-01"},
+            "children": [
+                {
+                    "name": "documents",
+                    "node_type": "folder",
+                    "children": [
+                        {
+                            "name": "report.pdf",
+                            "node_type": "file",
+                            "size_bytes": 1024,
+                            "permissions": [{"user": "alice", "level": "read"}],
+                        },
+                        {
+                            "name": "data",
+                            "node_type": "folder",
+                            "children": [
+                                {
+                                    "name": "archive.zip",
+                                    "node_type": "file",
+                                    "size_bytes": 2048,
+                                }
+                            ],
+                        },
+                    ],
+                },
+                {
+                    "name": "shortcut",
+                    "node_type": "symlink",
+                    "symlink_target": {
+                        "name": "target_file",
+                        "node_type": "file",
+                        "size_bytes": 512,
+                    },
+                },
+            ],
+        }
+
+        result = TruePartial.model_validate(data)
+        assert result.name == "root"
+        assert result.node_type == NodeType.FOLDER
+        assert len(result.children) == 2
+        assert result.children[0].name == "documents"
+        assert len(result.children[0].children) == 2
+        assert result.children[0].children[0].name == "report.pdf"
+        assert result.children[0].children[0].size_bytes == 1024
+        assert result.children[0].children[1].children[0].name == "archive.zip"
+        assert result.children[1].symlink_target.name == "target_file"
+        assert result.permissions[0].level == "admin"
+
+    def test_recursive_with_union_types(self):
+        """Recursive model with Union types containing self-references."""
+        from typing import Union
+
+        class TextBlock(BaseModel):
+            text: str
+
+        class Container(BaseModel):
+            title: str
+            content: list[Union[TextBlock, "Container"]]
+
+        Container.model_rebuild()
+
+        PartialContainer = Partial[Container]
+        TruePartial = PartialContainer.get_partial_model()
+
+        data = {
+            "title": "Chapter 1",
+            "content": [
+                {"text": "Introduction paragraph"},
+                {
+                    "title": "Section 1.1",
+                    "content": [
+                        {"text": "Section text"},
+                        {
+                            "title": "Subsection 1.1.1",
+                            "content": [{"text": "Deep nested text"}],
+                        },
+                    ],
+                },
+                {"text": "Closing paragraph"},
+            ],
+        }
+
+        result = TruePartial.model_validate(data)
+        assert result.title == "Chapter 1"
+        assert len(result.content) == 3
+        assert result.content[0].text == "Introduction paragraph"
+        assert result.content[1].title == "Section 1.1"
+        assert result.content[1].content[1].title == "Subsection 1.1.1"