Fix $ref handling in request-response mode for readOnly/writeOnly schemas

Author: koxudaxi

src/datamodel_code_generator/parser/jsonschema.py

@@ -973,12 +973,139 @@ def _should_generate_separate_models(
 
     def _should_generate_base_model(self, *, generates_separate_models: bool = False) -> bool:
         """Determine if Base model should be generated."""
+        if getattr(self, "_force_base_model_generation", False):
+            return True
         if self.read_only_write_only_model_type is None:
             return True
         if self.read_only_write_only_model_type == ReadOnlyWriteOnlyModelType.All:
             return True
         return not generates_separate_models
 
+    def _ref_schema_generates_variant(self, ref_path: str, suffix: str) -> bool:
+        """Check if a referenced schema will generate a specific variant (Request or Response).
+
+        For Request variant: schema must have readOnly fields AND at least one non-readOnly field.
+        For Response variant: schema must have writeOnly fields AND at least one non-writeOnly field.
+        """
+        try:
+            ref_schema = self._load_ref_schema_object(ref_path)
+        except Exception:  # noqa: BLE001  # pragma: no cover
+            return False
+
+        has_read_only = False
+        has_write_only = False
+        has_non_read_only = False
+        has_non_write_only = False
+
+        for prop in (ref_schema.properties or {}).values():
+            if not isinstance(prop, JsonSchemaObject):
+                continue
+            is_read_only = self._resolve_field_flag(prop, "readOnly")
+            is_write_only = self._resolve_field_flag(prop, "writeOnly")
+            if is_read_only:
+                has_read_only = True
+            else:
+                has_non_read_only = True
+            if is_write_only:
+                has_write_only = True
+            else:
+                has_non_write_only = True
+
+        if suffix == "Request":
+            return has_read_only and has_non_read_only
+        if suffix == "Response":
+            return has_write_only and has_non_write_only
+        return False  # pragma: no cover
+
+    def _ref_schema_has_model(self, ref_path: str) -> bool:
+        """Check if a referenced schema will have a model (base or variant) generated.
+
+        Returns False if the schema has only readOnly or only writeOnly fields in request-response mode,
+        which would result in no model being generated at all.
+        """
+        try:
+            ref_schema = self._load_ref_schema_object(ref_path)
+        except Exception:  # noqa: BLE001  # pragma: no cover
+            return True
+
+        has_read_only = False
+        has_write_only = False
+
+        for prop in (ref_schema.properties or {}).values():
+            if not isinstance(prop, JsonSchemaObject):
+                continue
+            is_read_only = self._resolve_field_flag(prop, "readOnly")
+            is_write_only = self._resolve_field_flag(prop, "writeOnly")
+            if is_read_only:
+                has_read_only = True
+            elif is_write_only:
+                has_write_only = True
+            else:  # pragma: no cover
+                return True
+
+        if has_read_only and not has_write_only:
+            return False
+        return not (has_write_only and not has_read_only)
+
+    def _update_data_type_ref_for_variant(self, data_type: DataType, suffix: str) -> None:
+        """Recursively update data type references to point to variant models."""
+        if data_type.reference:
+            ref_path = data_type.reference.path
+            if self._ref_schema_generates_variant(ref_path, suffix):
+                path_parts = ref_path.split("/")
+                base_name = path_parts[-1]
+                variant_name = f"{base_name}{suffix}"
+                unique_name = self.model_resolver.get_class_name(variant_name, unique=False).name
+                path_parts[-1] = unique_name
+                variant_ref = self.model_resolver.add(path_parts, unique_name, class_name=True, unique=False)
+                data_type.reference = variant_ref
+            elif not self._ref_schema_has_model(ref_path):
+                if not hasattr(self, "_force_base_model_refs"):
+                    self._force_base_model_refs: set[str] = set()
+                self._force_base_model_refs.add(ref_path)
+        for nested_dt in data_type.data_types:
+            self._update_data_type_ref_for_variant(nested_dt, suffix)
+
+    def _update_field_refs_for_variant(
+        self, model_fields: list[DataModelFieldBase], suffix: str
+    ) -> list[DataModelFieldBase]:
+        """Update field references in model_fields to point to variant models.
+
+        For Request models, refs should point to Request variants.
+        For Response models, refs should point to Response variants.
+        """
+        if self.read_only_write_only_model_type != ReadOnlyWriteOnlyModelType.RequestResponse:
+            return model_fields
+        for field in model_fields:
+            if field.data_type:
+                self._update_data_type_ref_for_variant(field.data_type, suffix)
+        return model_fields
+
+    def _generate_forced_base_models(self) -> None:
+        """Generate base models for schemas that are referenced as property types but lack models."""
+        if not hasattr(self, "_force_base_model_refs"):
+            return
+        if not self._force_base_model_refs:  # pragma: no cover
+            return
+
+        existing_model_paths = {result.path for result in self.results}
+
+        for ref_path in sorted(self._force_base_model_refs):
+            if ref_path in existing_model_paths:  # pragma: no cover
+                continue
+            try:
+                ref_schema = self._load_ref_schema_object(ref_path)
+                path_parts = ref_path.split("/")
+                schema_name = path_parts[-1]
+
+                self._force_base_model_generation = True
+                try:
+                    self.parse_obj(schema_name, ref_schema, path_parts)
+                finally:
+                    self._force_base_model_generation = False
+            except Exception:  # noqa: BLE001, S110  # pragma: no cover
+                pass
+
     def _create_variant_model(  # noqa: PLR0913, PLR0917
         self,
         path: list[str],
@@ -991,6 +1118,8 @@ def _create_variant_model(  # noqa: PLR0913, PLR0917
         """Create a Request or Response model variant."""
         if not model_fields:
             return
+        # Update field refs to point to variant models when in request-response mode
+        self._update_field_refs_for_variant(model_fields, suffix)
         variant_name = f"{base_name}{suffix}"
         unique_name = self.model_resolver.get_class_name(variant_name, unique=True).name
         model_path = [*path[:-1], unique_name]
@@ -3720,6 +3849,7 @@ def parse_raw(self) -> None:
             self._parse_file(self.raw_obj, obj_name, path_parts)
 
         self._resolve_unparsed_json_pointer()
+        self._generate_forced_base_models()
 
     def _resolve_unparsed_json_pointer(self) -> None:
         """Resolve any remaining unparsed JSON pointer references recursively."""

src/datamodel_code_generator/parser/openapi.py

@@ -794,6 +794,7 @@ def parse_raw(self) -> None:  # noqa: PLR0912
                         )
 
         self._resolve_unparsed_json_pointer()
+        self._generate_forced_base_models()
 
     def _collect_discriminator_schemas(self) -> None:
         """Collect schemas with discriminators but no oneOf/anyOf, and find their subtypes."""

tests/data/expected/main/openapi/read_only_write_only_ref_request_response.py

@@ -0,0 +1,35 @@
+# generated by datamodel-codegen:
+#   filename:  read_only_write_only_ref_request_response.yaml
+#   timestamp: 2019-07-26T00:00:00+00:00
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class ChildMixedRequest(BaseModel):
+    name: str | None = None
+
+
+class ParentWithMixedChildRequest(BaseModel):
+    child: ChildMixedRequest | None = None
+
+
+class ParentWithChildListRequest(BaseModel):
+    children: list[ChildMixedRequest] | None = None
+
+
+class ChildOnlyReadOnly(BaseModel):
+    id: int | None = None
+
+
+class ChildOnlyWriteOnly(BaseModel):
+    secret: str | None = None
+
+
+class ParentWithOnlyReadOnlyChildRequest(BaseModel):
+    child: ChildOnlyReadOnly | None = None
+
+
+class ParentWithOnlyWriteOnlyChildResponse(BaseModel):
+    child: ChildOnlyWriteOnly | None = None

tests/data/openapi/read_only_write_only_ref_request_response.yaml

@@ -0,0 +1,68 @@
+openapi: "3.0.0"
+info:
+  title: Read Only Write Only Ref Request Response Test API
+  version: "1.0"
+paths: {}
+components:
+  schemas:
+    # Child with only readOnly fields - should generate base model when referenced directly
+    ChildOnlyReadOnly:
+      type: object
+      properties:
+        id:
+          readOnly: true
+          type: integer
+    # Child with only writeOnly fields - should generate base model when referenced directly
+    ChildOnlyWriteOnly:
+      type: object
+      properties:
+        secret:
+          writeOnly: true
+          type: string
+    # Child with mixed fields - should generate ChildRequest variant
+    ChildMixed:
+      type: object
+      properties:
+        id:
+          readOnly: true
+          type: integer
+        name:
+          type: string
+    # Parent referencing child with only readOnly fields
+    ParentWithOnlyReadOnlyChild:
+      type: object
+      properties:
+        child:
+          $ref: "#/components/schemas/ChildOnlyReadOnly"
+        name:
+          readOnly: true
+          type: string
+    # Parent referencing child with only writeOnly fields
+    ParentWithOnlyWriteOnlyChild:
+      type: object
+      properties:
+        child:
+          $ref: "#/components/schemas/ChildOnlyWriteOnly"
+        secret:
+          writeOnly: true
+          type: string
+    # Parent referencing child with mixed fields
+    ParentWithMixedChild:
+      type: object
+      properties:
+        child:
+          $ref: "#/components/schemas/ChildMixed"
+        name:
+          readOnly: true
+          type: string
+    # Parent with list of children (nested type reference)
+    ParentWithChildList:
+      type: object
+      properties:
+        children:
+          type: array
+          items:
+            $ref: "#/components/schemas/ChildMixed"
+        name:
+          readOnly: true
+          type: string

tests/main/openapi/test_main_openapi.py

@@ -4421,6 +4421,29 @@ def test_main_openapi_read_only_write_only_empty_base(output_file: Path) -> None
     )
 
 
+def test_main_openapi_read_only_write_only_ref_request_response(output_file: Path) -> None:
+    """Test readOnly/writeOnly with $ref in request-response mode (issue #2940).
+
+    When a schema references another schema via $ref:
+    - If the referenced schema generates variants, use the variant reference
+    - If the referenced schema would have no model (only readOnly/writeOnly fields),
+      force generation of the base model
+    """
+    run_main_and_assert(
+        input_path=OPEN_API_DATA_PATH / "read_only_write_only_ref_request_response.yaml",
+        output_path=output_file,
+        input_file_type="openapi",
+        assert_func=assert_file_content,
+        expected_file="read_only_write_only_ref_request_response.py",
+        extra_args=[
+            "--output-model-type",
+            "pydantic_v2.BaseModel",
+            "--read-only-write-only-model-type",
+            "request-response",
+        ],
+    )
+
+
 def test_main_openapi_dot_notation_inheritance(output_dir: Path) -> None:
     """Test dot notation in schema names with inheritance."""
     run_main_and_assert(