feat: add support for reserved names and improve error messaging for missing anchors (#2568)

Author: koxudaxi

src/datamodel_code_generator/reference.py

@@ -34,6 +34,7 @@
 from packaging import version
 from pydantic import BaseModel, Field
 
+from datamodel_code_generator import Error
 from datamodel_code_generator.util import PYDANTIC_V2, ConfigDict, model_validator
 
 if TYPE_CHECKING:
@@ -270,9 +271,10 @@ def get_valid_name(  # noqa: PLR0912
         else:
             new_name = name
         while (
-            not (new_name.isidentifier() or not self._validate_field_name(new_name))
+            not new_name.isidentifier()
             or iskeyword(new_name)
             or (excludes and new_name in excludes)
+            or not self._validate_field_name(new_name)
         ):
             new_name = f"{name}{count}" if upper_camel else f"{name}_{count}"
             count += 1
@@ -460,8 +462,6 @@ def base_url_context(self, base_url: str) -> Generator[None, None, None]:
     @property
     def current_root(self) -> Sequence[str]:
         """Return the current root path components."""
-        if len(self._current_root) > 1:
-            return self._current_root
         return self._current_root
 
     def set_current_root(self, current_root: Sequence[str]) -> None:
@@ -504,22 +504,29 @@ def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR09
             return f"{'/'.join(self.current_root)}#"
         if self.current_base_path and not self.base_url and joined_path[0] != "#" and not is_url(joined_path):
             # resolve local file path
-            file_path, *object_part = joined_path.split("#", 1)
+            file_path, fragment = joined_path.split("#", 1) if "#" in joined_path else (joined_path, "")
             resolved_file_path = Path(self.current_base_path, file_path).resolve()
             joined_path = get_relative_path(self._base_path, resolved_file_path).as_posix()
-            if object_part:
-                joined_path += f"#{object_part[0]}"
+            if fragment:
+                joined_path += f"#{fragment}"
         if ID_PATTERN.match(joined_path):
-            ref: str = self.ids["/".join(self.current_root)][joined_path]
+            id_scope = "/".join(self.current_root)
+            scoped_ids = self.ids[id_scope]
+            ref: str | None = scoped_ids.get(joined_path)
+            if ref is None:
+                msg = (
+                    f"Unresolved $id reference '{joined_path}' in scope '{id_scope or '<root>'}'. "
+                    f"Known $id values: {', '.join(sorted(scoped_ids)) or '<none>'}"
+                )
+                raise Error(msg)
         else:
             if "#" not in joined_path:
                 joined_path += "#"
             elif joined_path[0] == "#":
                 joined_path = f"{'/'.join(self.current_root)}{joined_path}"
 
-            delimiter = joined_path.index("#")
-            file_path = "".join(joined_path[:delimiter])
-            ref = f"{''.join(joined_path[:delimiter])}#{''.join(joined_path[delimiter + 1 :])}"
+            file_path, fragment = joined_path.split("#", 1)
+            ref = f"{file_path}#{fragment}"
             if self.root_id_base_path and not (is_url(joined_path) or Path(self._base_path, file_path).is_file()):
                 ref = f"{self.root_id_base_path}/{ref}"
 
@@ -544,8 +551,11 @@ def resolve_ref(self, path: Sequence[str] | str) -> str:  # noqa: PLR0911, PLR09
                 root_id_url.netloc,
             ):  # pragma: no cover
                 target_url_path = Path(target_url.path)
-                relative_target_base = get_relative_path(Path(root_id_url.path).parent, target_url_path.parent)
-                target_path = self.current_base_path / relative_target_base / target_url_path.name
+                target_path = (
+                    self.current_base_path
+                    / get_relative_path(Path(root_id_url.path).parent, target_url_path.parent)
+                    / target_url_path.name
+                )
                 if target_path.exists():
                     return f"{target_path.resolve().relative_to(self._base_path)}#{path_part}"
 

tests/data/expected/main/jsonschema/reserved_property.py

@@ -0,0 +1,13 @@
+# generated by datamodel-codegen:
+#   filename:  reserved_property.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class ReservedNames(BaseModel):
+    json_: str = Field(..., alias='json')
+    schema_: str = Field(..., alias='schema')
+    dict_: int = Field(..., alias='dict')

tests/data/expected/main/jsonschema/with_anchor.py

@@ -0,0 +1,15 @@
+# generated by datamodel-codegen:
+#   filename:  with_anchor.json
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Address(BaseModel):
+    street: str
+
+
+class Model(BaseModel):
+    billing_address: Address

tests/data/jsonschema/missing_anchor.json

@@ -0,0 +1,5 @@
+{
+  "title": "BrokenAnchor",
+  "type": "object",
+  "$ref": "#address"
+}

tests/data/jsonschema/reserved_property.json

@@ -0,0 +1,10 @@
+{
+  "title": "ReservedNames",
+  "type": "object",
+  "properties": {
+    "json": {"type": "string"},
+    "schema": {"type": "string"},
+    "dict": {"type": "integer"}
+  },
+  "required": ["json", "schema", "dict"]
+}

tests/data/jsonschema/with_anchor.json

@@ -0,0 +1,18 @@
+{
+  "$id": "http://example.com/root.json",
+  "type": "object",
+  "definitions": {
+    "Address": {
+      "$id": "#address",
+      "type": "object",
+      "properties": {
+        "street": { "type": "string" }
+      },
+      "required": ["street"]
+    }
+  },
+  "properties": {
+    "billing_address": { "$ref": "#address" }
+  },
+  "required": ["billing_address"]
+}

tests/main/jsonschema/test_main_jsonschema.py

@@ -374,6 +374,40 @@ def test_main_invalid_model_name(output_file: Path) -> None:
     )
 
 
+def test_main_jsonschema_reserved_field_names(output_file: Path) -> None:
+    """Test reserved names are safely suffixed and aliased."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "reserved_property.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="reserved_property.py",
+    )
+
+
+def test_main_jsonschema_with_local_anchor(output_file: Path) -> None:
+    """Test $id anchor lookup resolves without error and reuses definitions."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "with_anchor.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="with_anchor.py",
+    )
+
+
+def test_main_jsonschema_missing_anchor_reports_error(capsys: pytest.CaptureFixture[str], output_file: Path) -> None:
+    """Test missing $id anchor produces a clear error instead of KeyError trace."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "missing_anchor.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        expected_exit=Exit.ERROR,
+        capsys=capsys,
+        expected_stderr_contains="Unresolved $id reference '#address'",
+    )
+
+
 def test_main_root_id_jsonschema_with_local_file(mocker: MockerFixture, output_file: Path) -> None:
     """Test root ID JSON Schema with local file reference."""
     root_id_response = mocker.Mock()