Handle nested schema namespaces

src/datamodel_code_generator/parser/jsonschema.py

@@ -5443,15 +5443,25 @@ def _handle_python_import(
     def _is_named_schema_definition_path(self, path: list[str]) -> bool:
         """Check if path points to a named schema entry under definitions/$defs."""
         current_root = list(self.model_resolver.current_root)
-        expected_path_length = len(current_root) + 2
-        if len(path) != expected_path_length:
+        if len(path) < len(current_root) + 2:
             return False
 
         schema_container_path = path[len(current_root)]
         return path[: len(current_root)] == current_root and any(
             schema_container_path == schema_path for schema_path, _ in self.schema_paths
         )
 
+    def _is_current_root_schema_path(self, path: list[str]) -> bool:
+        current_root = list(self.model_resolver.current_root)
+        if path == (current_root or ["#"]):
+            return True
+        return self.model_resolver.resolve_ref(path) == self.model_resolver.resolve_ref(current_root or "#")
+
+    def _drop_ref_from_schema(self, obj: JsonSchemaObject) -> JsonSchemaObject:
+        return self.SCHEMA_OBJECT_TYPE.model_validate(
+            obj.model_dump(exclude={"ref"}, exclude_unset=True, by_alias=True)
+        )
+
     def parse_obj(  # noqa: PLR0912
         self,
         name: str,
@@ -5460,7 +5470,10 @@ def parse_obj(  # noqa: PLR0912
     ) -> None:
         """Parse a JsonSchemaObject by dispatching to appropriate parse methods."""
         if obj.has_ref_with_schema_keywords and not obj.is_ref_with_nullable_only:
-            obj = self._merge_ref_with_schema(obj)
+            if obj.ref == "#" and self._is_current_root_schema_path(path):
+                obj = self._drop_ref_from_schema(obj)
+            else:
+                obj = self._merge_ref_with_schema(obj)
             if obj.ref:
                 if self._is_named_schema_definition_path(path):
                     self.parse_root_type(name, obj, path)
@@ -5646,6 +5659,78 @@ def parse_json_pointer(self, raw: dict[str, YamlValue], ref: str, path_parts: li
 
         self.parse_raw_obj(model_name, models, [*path_parts, f"#/{reference_paths[0]}", *reference_paths[1:]])
 
+    def _known_schema_object_raw_keys(self) -> set[str]:
+        keys = {"definitions", "$defs"}
+        for name, field in self.SCHEMA_OBJECT_TYPE.get_fields().items():  # ty: ignore
+            keys.add(name)
+            if alias := getattr(field, "alias", None):
+                keys.add(alias)
+        return keys
+
+    def _has_schema_affecting_keywords(self, raw: dict[str, Any]) -> bool:
+        metadata_keys = {
+            *self.SCHEMA_OBJECT_TYPE.__metadata_only_fields__,
+            "extras",
+            self.SCHEMA_OBJECT_TYPE.__extra_key__,
+        }
+        schema_affecting_keys = {
+            *self._known_schema_object_raw_keys(),
+            *self.SCHEMA_OBJECT_TYPE.__schema_affecting_extras__,
+        } - metadata_keys
+        return any(str(key) in schema_affecting_keys for key in raw)
+
+    def _is_version_definition_namespace_name(self, name: str) -> bool:  # noqa: PLR6301
+        return re.fullmatch(r"v\d+(?:[._-]\d+)*", name, flags=re.IGNORECASE) is not None
+
+    def _iter_definition_namespace_entries(
+        self,
+        raw: dict[str, Any],
+        path: list[str],
+        *,
+        include_direct_children: bool,
+    ) -> Iterator[tuple[str, YamlValue, list[str]]]:
+        for schema_key in ("definitions", "$defs"):
+            if isinstance(definitions := raw.get(schema_key), dict):
+                yield from self._iter_schema_definition_entries(definitions, [*path, schema_key])
+
+        if not include_direct_children:
+            return
+
+        known_keys = self._known_schema_object_raw_keys()
+        for key, value in raw.items():
+            key_str = str(key)
+            if key_str in known_keys or key_str.startswith("x-") or not isinstance(value, (dict, bool)):
+                continue
+            yield from self._iter_schema_definition_entry(key_str, value, [*path, key_str])
+
+    def _iter_schema_definition_entry(
+        self,
+        name: str,
+        raw: YamlValue,
+        path: list[str],
+    ) -> Iterator[tuple[str, YamlValue, list[str]]]:
+        if isinstance(raw, dict) and not self._has_schema_affecting_keywords(raw):
+            entries = list(
+                self._iter_definition_namespace_entries(
+                    raw,
+                    path,
+                    include_direct_children=self._is_version_definition_namespace_name(name),
+                )
+            )
+            if entries:
+                yield from entries
+                return
+        yield name, raw, path
+
+    def _iter_schema_definition_entries(
+        self,
+        definitions: dict[str, YamlValue],
+        base_path: list[str],
+    ) -> Iterator[tuple[str, YamlValue, list[str]]]:
+        for key, model in definitions.items():
+            name = str(key)
+            yield from self._iter_schema_definition_entry(name, model, [*base_path, name])
+
     def _parse_file(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
         self,
         raw: dict[str, Any],
@@ -5697,8 +5782,16 @@ def _parse_file(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
                     except KeyError:  # pragma: no cover
                         continue
 
-                for key, model in definitions.items():
-                    definition_path = [*path_parts, schema_path, key]
+                definition_entries = list(self._iter_schema_definition_entries(definitions, [*path_parts, schema_path]))
+                definition_metadata_entries = [
+                    *((str(key), model, [*path_parts, schema_path, str(key)]) for key, model in definitions.items()),
+                    *definition_entries,
+                ]
+                seen_definition_metadata_paths: set[tuple[str, ...]] = set()
+                for _key, model, definition_path in definition_metadata_entries:
+                    if (definition_path_key := tuple(definition_path)) in seen_definition_metadata_paths:
+                        continue
+                    seen_definition_metadata_paths.add(definition_path_key)
                     obj = self._validate_schema_object(model, definition_path)
                     self.parse_id(obj, definition_path)
                     if obj.recursiveAnchor:
@@ -5714,8 +5807,7 @@ def _parse_file(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
                     self.parse_obj(model_name, self._validate_schema_object(models, path), path)
                 elif not self.skip_root_model:
                     self.parse_obj(obj_name, root_obj, path_parts or ["#"])
-                for key, model in definitions.items():
-                    path = [*path_parts, schema_path, key]
+                for key, model, path in definition_entries:
                     reference = self.model_resolver.get(path)
                     if not reference or not reference.loaded:
                         self.parse_raw_obj(key, model, path)

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

@@ -0,0 +1,18 @@
+# generated by datamodel-codegen:
+#   filename:  definitions_v2_namespace.json
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class User(BaseModel):
+    id: int | None = None
+
+
+class Audit(BaseModel):
+    event: str | None = None
+
+
+class Envelope(BaseModel):
+    user: User | None = None

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

@@ -0,0 +1,19 @@
+# generated by datamodel-codegen:
+#   filename:  nested_defs_namespace_ref.json
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Pet(BaseModel):
+    tag: str | None = None
+
+
+class User(BaseModel):
+    name: str | None = None
+
+
+class Envelope(BaseModel):
+    user: User | None = None
+    pet: Pet | None = None

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

@@ -0,0 +1,10 @@
+# generated by datamodel-codegen:
+#   filename:  root_ref_self_with_keywords.json
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Model(BaseModel):
+    name: str

tests/data/jsonschema/definitions_v2_namespace.json

@@ -0,0 +1,30 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "Envelope",
+  "type": "object",
+  "properties": {
+    "user": {
+      "$ref": "#/definitions/v2/User"
+    }
+  },
+  "definitions": {
+    "v2": {
+      "User": {
+        "type": "object",
+        "properties": {
+          "id": {
+            "type": "integer"
+          }
+        }
+      },
+      "Audit": {
+        "type": "object",
+        "properties": {
+          "event": {
+            "type": "string"
+          }
+        }
+      }
+    }
+  }
+}

tests/data/jsonschema/nested_defs_namespace_ref.json

@@ -0,0 +1,37 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "Envelope",
+  "type": "object",
+  "properties": {
+    "user": {
+      "$ref": "#/$defs/v2/$defs/User"
+    },
+    "pet": {
+      "$ref": "#/$defs/v2/definitions/Pet"
+    }
+  },
+  "$defs": {
+    "v2": {
+      "$defs": {
+        "User": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            }
+          }
+        }
+      },
+      "definitions": {
+        "Pet": {
+          "type": "object",
+          "properties": {
+            "tag": {
+              "type": "string"
+            }
+          }
+        }
+      }
+    }
+  }
+}

tests/data/jsonschema/root_ref_self_with_keywords.json

@@ -0,0 +1,11 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$ref": "#",
+  "type": "object",
+  "required": ["name"],
+  "properties": {
+    "name": {
+      "type": "string"
+    }
+  }
+}

tests/main/jsonschema/test_main_jsonschema.py

@@ -2035,6 +2035,42 @@ def test_main_nested_json_pointer(output_file: Path) -> None:
     )
 
 
+def test_main_jsonschema_definitions_namespace(output_file: Path) -> None:
+    """Test nested namespace-like definitions are parsed without wrapper models."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "definitions_v2_namespace.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="definitions_v2_namespace.py",
+        extra_args=["--disable-timestamp"],
+    )
+
+
+def test_main_jsonschema_nested_defs_namespace_ref(output_file: Path) -> None:
+    """Test nested $defs/definitions namespace references skip wrapper models."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "nested_defs_namespace_ref.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="nested_defs_namespace_ref.py",
+        extra_args=["--disable-timestamp"],
+    )
+
+
+def test_main_jsonschema_root_ref_self_with_keywords(output_file: Path) -> None:
+    """Test root $ref '#' with local schema keywords parses the local schema."""
+    run_main_and_assert(
+        input_path=JSON_SCHEMA_DATA_PATH / "root_ref_self_with_keywords.json",
+        output_path=output_file,
+        input_file_type="jsonschema",
+        assert_func=assert_file_content,
+        expected_file="root_ref_self_with_keywords.py",
+        extra_args=["--disable-timestamp"],
+    )
+
+
 def test_main_jsonschema_multiple_files_json_pointer(output_dir: Path) -> None:
     """Test JSON pointer with multiple files."""
     run_main_and_assert(

tests/parser/test_jsonschema.py

@@ -1862,6 +1862,27 @@ def test_jsonschema_parser_edge_case_helpers() -> None:
     assert parser._get_data_type_from_json_value(object()).type_hint == "Any"
 
 
+@pytest.mark.parametrize(
+    ("current_root", "path", "expected"),
+    [
+        ([], [], True),
+        ([], ["#"], True),
+        (["schema.json"], ["schema.json"], True),
+        (["schema.json"], ["schema.json", "#"], True),
+        (["schema.json"], ["schema.json#"], True),
+        (["schema.json"], ["other.json#"], False),
+    ],
+)
+def test_is_current_root_schema_path_normalizes_root_spellings(
+    current_root: list[str], path: list[str], *, expected: bool
+) -> None:
+    """Treat equivalent root path spellings as the current schema root."""
+    parser = JsonSchemaParser("")
+    parser.model_resolver.set_current_root(current_root)
+
+    assert parser._is_current_root_schema_path(path) is expected
+
+
 def test_anchor_ref_path_escapes_json_pointer_segments() -> None:
     """Test anchor ref paths escape JSON Pointer segments."""
     parser = JsonSchemaParser("")