fix: use alias for shadowing pydantic fields in json schema

Author: ionut-mihalache-uipath

src/uipath_langchain/agent/react/jsonschema_pydantic_converter.py

@@ -13,6 +13,12 @@
 # This allows get_type_hints() to resolve forward references
 _DYNAMIC_MODULE_NAME = "jsonschema_pydantic_converter._dynamic"
 
+# Field names that shadow BaseModel attributes and must be renamed.
+# Computed from BaseModel's public interface to stay future-proof across Pydantic versions.
+_RESERVED_FIELD_NAMES: frozenset[str] = frozenset(
+    name for name in dir(BaseModel) if not name.startswith("_")
+)
+
 
 def _get_or_create_dynamic_module() -> ModuleType:
     """Get or create the shared pseudo-module for dynamic types."""
@@ -25,6 +31,116 @@ def _get_or_create_dynamic_module() -> ModuleType:
     return sys.modules[_DYNAMIC_MODULE_NAME]
 
 
+def _rename_reserved_properties(
+    schema: dict[str, Any],
+) -> tuple[dict[str, Any], dict[str, str]]:
+    """Rename JSON Schema properties that would shadow BaseModel attributes.
+
+    Appends '_' to reserved property names in a copy of the schema. Tracks all
+    renames so aliases can be applied to the resulting Pydantic model.
+
+    Returns:
+        Tuple of (modified schema copy, {new_field_name: original_name}).
+    """
+    renames: dict[str, str] = {}
+
+    def _process(s: dict[str, Any]) -> dict[str, Any]:
+        result = s.copy()
+
+        if "properties" in result:
+            existing_keys = set(result["properties"].keys())
+            new_props: dict[str, Any] = {}
+
+            for key, value in result["properties"].items():
+                if key in _RESERVED_FIELD_NAMES:
+                    new_key = key + "_"
+                    while new_key in existing_keys:
+                        new_key += "_"
+                    renames[new_key] = key
+                else:
+                    new_key = key
+
+                new_props[new_key] = (
+                    _process(value) if isinstance(value, dict) else value
+                )
+            result["properties"] = new_props
+
+            if "required" in result:
+                # Build a lookup from original→renamed for this level only
+                local_renames = {v: k for k, v in renames.items() if v in existing_keys}
+                result["required"] = [
+                    local_renames.get(name, name) for name in result["required"]
+                ]
+
+        for defs_key in ("$defs", "definitions"):
+            if defs_key in result:
+                result[defs_key] = {
+                    k: (_process(v) if isinstance(v, dict) else v)
+                    for k, v in result[defs_key].items()
+                }
+
+        if "items" in result and isinstance(result["items"], dict):
+            result["items"] = _process(result["items"])
+
+        for keyword in ("allOf", "anyOf", "oneOf"):
+            if keyword in result:
+                result[keyword] = [
+                    _process(sub) if isinstance(sub, dict) else sub
+                    for sub in result[keyword]
+                ]
+
+        if "not" in result and isinstance(result["not"], dict):
+            result["not"] = _process(result["not"])
+
+        for keyword in ("if", "then", "else"):
+            if keyword in result and isinstance(result[keyword], dict):
+                result[keyword] = _process(result[keyword])
+
+        return result
+
+    modified = _process(schema)
+    return modified, renames
+
+
+def _apply_field_aliases(
+    model: Type[BaseModel],
+    namespace: dict[str, Any],
+    renames: dict[str, str],
+) -> None:
+    """Add aliases to renamed fields so serialization/validation uses original names.
+
+    Iterates the root model and all nested models from the namespace. For any
+    field whose name appears in ``renames``, sets alias/validation_alias/
+    serialization_alias to the original property name and enables
+    ``populate_by_name`` + ``serialize_by_alias`` in the model config.
+    """
+    if not renames:
+        return
+
+    all_models = [model]
+    for v in namespace.values():
+        if inspect.isclass(v) and issubclass(v, BaseModel):
+            all_models.append(v)
+
+    for m in all_models:
+        needs_rebuild = False
+        for field_name, field_info in m.model_fields.items():
+            if field_name in renames:
+                original_name = renames[field_name]
+                field_info.alias = original_name
+                field_info.validation_alias = original_name
+                field_info.serialization_alias = original_name
+                needs_rebuild = True
+
+        if needs_rebuild:
+            m.model_config = {
+                **m.model_config,
+                "populate_by_name": True,
+                "serialize_by_alias": True,
+            }
+            m.model_rebuild(force=True)
+
+
 def create_model(
     schema: dict[str, Any],
 ) -> Type[BaseModel]:
@@ -36,7 +152,9 @@ def create_model(
             category=UiPathErrorCategory.USER,
         )
 
-    model, namespace = transform_with_modules(schema)
+    processed_schema, renames = _rename_reserved_properties(schema)
+    model, namespace = transform_with_modules(processed_schema)
+    _apply_field_aliases(model, namespace, renames)
     corrected_namespace: dict[str, Any] = {}
 
     def collect_types(annotation: Any) -> None:

tests/agent/react/test_schemas.py

@@ -3,9 +3,12 @@
 from typing import Any
 
 import pytest
+from pydantic import BaseModel
 
 from uipath_langchain.agent.exceptions import AgentStartupError, AgentStartupErrorCode
 from uipath_langchain.agent.react.jsonschema_pydantic_converter import (
+    _RESERVED_FIELD_NAMES,
+    _rename_reserved_properties,
     create_model,
     has_underscore_fields,
 )
@@ -309,3 +312,224 @@ def test_nested_underscore_field(self) -> None:
         assert exc_info.value.error_info.code == AgentStartupError.full_code(
             AgentStartupErrorCode.UNDERSCORE_SCHEMA
         )
+
+
+class TestRenameReservedProperties:
+    """Tests for _rename_reserved_properties schema pre-processing."""
+
+    def test_renames_schema_field(self) -> None:
+        schema = {
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "name": {"type": "string"},
+            },
+            "required": ["schema", "name"],
+        }
+        modified, renames = _rename_reserved_properties(schema)
+
+        assert "schema_" in modified["properties"]
+        assert "schema" not in modified["properties"]
+        assert "name" in modified["properties"]
+        assert renames == {"schema_": "schema"}
+        assert modified["required"] == ["schema_", "name"]
+
+    def test_renames_multiple_reserved_fields(self) -> None:
+        schema = {
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "copy": {"type": "string"},
+                "validate": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        modified, renames = _rename_reserved_properties(schema)
+
+        assert "schema_" in modified["properties"]
+        assert "copy_" in modified["properties"]
+        assert "validate_" in modified["properties"]
+        assert "name" in modified["properties"]
+        assert len(renames) == 3
+
+    def test_handles_collision_with_existing_field(self) -> None:
+        """When 'schema_' already exists, 'schema' should become 'schema__'."""
+        schema = {
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "schema_": {"type": "string"},
+            },
+        }
+        modified, renames = _rename_reserved_properties(schema)
+
+        assert "schema__" in modified["properties"]
+        assert "schema_" in modified["properties"]
+        assert renames["schema__"] == "schema"
+
+    def test_renames_in_defs(self) -> None:
+        schema = {
+            "type": "object",
+            "properties": {"name": {"type": "string"}},
+            "$defs": {
+                "Inner": {
+                    "type": "object",
+                    "properties": {
+                        "schema": {"type": "string"},
+                    },
+                    "required": ["schema"],
+                },
+            },
+        }
+        modified, renames = _rename_reserved_properties(schema)
+
+        inner = modified["$defs"]["Inner"]
+        assert "schema_" in inner["properties"]
+        assert inner["required"] == ["schema_"]
+
+    def test_does_not_modify_original_schema(self) -> None:
+        schema = {
+            "type": "object",
+            "properties": {"schema": {"type": "string"}},
+        }
+        _rename_reserved_properties(schema)
+
+        assert "schema" in schema["properties"]
+
+    def test_no_renames_for_normal_fields(self) -> None:
+        schema = {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string"},
+                "age": {"type": "integer"},
+            },
+        }
+        modified, renames = _rename_reserved_properties(schema)
+
+        assert renames == {}
+        assert modified["properties"] == schema["properties"]
+
+
+class TestCreateModelWithReservedFields:
+    """Tests for create_model handling of reserved field names."""
+
+    def test_schema_field_creates_valid_model(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        assert issubclass(model, BaseModel)
+        # BaseModel methods should still work
+        assert callable(model.model_json_schema)
+        assert callable(model.model_validate)
+
+    def test_model_validate_accepts_original_names(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        instance = model.model_validate({"schema": "test_val", "name": "alice"})
+        assert instance is not None
+
+    def test_model_dump_outputs_original_names(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        instance = model.model_validate({"schema": "test_val", "name": "alice"})
+        dumped = instance.model_dump()
+
+        assert dumped == {"schema": "test_val", "name": "alice"}
+
+    def test_model_dump_json_mode_outputs_original_names(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        instance = model.model_validate({"schema": "val"})
+        dumped = instance.model_dump(mode="json")
+
+        assert dumped == {"schema": "val"}
+
+    def test_model_json_schema_shows_original_names(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "copy": {"type": "integer"},
+            },
+        }
+        model = create_model(schema)
+
+        json_schema = model.model_json_schema()
+        assert "schema" in json_schema["properties"]
+        assert "copy" in json_schema["properties"]
+        assert "schema_" not in json_schema["properties"]
+        assert "copy_" not in json_schema["properties"]
+
+    def test_multiple_reserved_fields(self) -> None:
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "schema": {"type": "string"},
+                "copy": {"type": "string"},
+                "validate": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        instance = model.model_validate(
+            {"schema": "s", "copy": "c", "validate": "v", "name": "n"}
+        )
+        dumped = instance.model_dump()
+
+        assert dumped == {"schema": "s", "copy": "c", "validate": "v", "name": "n"}
+
+    def test_model_fields_field_does_not_shadow(self) -> None:
+        """'model_fields' is in Pydantic's protected namespace — must not crash."""
+        schema = {
+            "title": "Input",
+            "type": "object",
+            "properties": {
+                "model_fields": {"type": "string"},
+                "name": {"type": "string"},
+            },
+        }
+        model = create_model(schema)
+
+        # model.model_fields should still be the Pydantic descriptor, not a field value
+        assert isinstance(model.model_fields, dict)
+        instance = model.model_validate({"model_fields": "test", "name": "n"})
+        assert instance.model_dump() == {"model_fields": "test", "name": "n"}
+
+    def test_reserved_field_names_constant_contains_known_problematic_names(
+        self,
+    ) -> None:
+        known_problematic = {"schema", "copy", "validate", "dict", "json", "construct"}
+        assert known_problematic.issubset(_RESERVED_FIELD_NAMES)