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

Author: koxudaxi

src/datamodel_code_generator/parser/jsonschema.py

@@ -979,6 +979,117 @@ def _should_generate_base_model(self, *, generates_separate_models: bool = False
             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
+
+        # Collect all fields from the schema
+        has_read_only = False
+        has_write_only = False
+        has_non_read_only = False
+        has_non_write_only = False
+
+        if ref_schema.properties:
+            for prop in ref_schema.properties.values():
+                if isinstance(prop, JsonSchemaObject):
+                    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
+
+        # For Request variant: needs readOnly fields AND at least one non-readOnly field remaining
+        if suffix == "Request":
+            return has_read_only and has_non_read_only
+        # For Response variant: needs writeOnly fields AND at least one non-writeOnly field remaining
+        if suffix == "Response":
+            return has_write_only and has_non_write_only
+        return False  # pragma: no cover - only Request and Response suffixes are used
+
+    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  # Assume it exists if we can't load it
+
+        # Check if the schema has any fields that aren't readOnly or writeOnly
+        has_read_only = False
+        has_write_only = False
+
+        if ref_schema.properties:
+            for prop in ref_schema.properties.values():
+                if isinstance(prop, JsonSchemaObject):
+                    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 - plain fields in refs to non-variant schemas
+                        # Has a plain field - schema will have a model
+                        return True
+
+        # Has ONLY readOnly fields - no model in request-response mode
+        if has_read_only and not has_write_only:
+            return False
+        # Has ONLY writeOnly fields - no model, or has both/neither - has model
+        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):
+                # The referenced schema will have a variant model - update the ref
+                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
+                # Create the variant path (e.g., "#/components/schemas/ChildRequest")
+                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):
+                # The referenced schema won't have any model - mark it for forced base model generation
+                if not hasattr(self, "_force_base_model_refs"):
+                    self._force_base_model_refs: set[str] = set()
+                self._force_base_model_refs.add(ref_path)
+        # Recursively update nested data types (for List[Child], Optional[Child], etc.)
+        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 _create_variant_model(  # noqa: PLR0913, PLR0917
         self,
         path: list[str],
@@ -991,6 +1102,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]

src/datamodel_code_generator/parser/openapi.py

@@ -699,8 +699,50 @@ def parse_operation(
                 path=[*path, "tags"],
             )
 
-    def parse_raw(self) -> None:  # noqa: PLR0912
+    def parse_raw(self) -> None:
         """Parse OpenAPI specification including schemas, paths, and operations."""
+        self._parse_raw_impl()
+        # Post-process: generate base models for schemas that are referenced as property types
+        # but would otherwise have no model in request-response mode
+        self._generate_forced_base_models()
+
+    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
+
+        # Get the set of schema paths that already have models
+        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  # Already has a model
+            try:
+                ref_schema = self._load_ref_schema_object(ref_path)
+                # Get the name from the path
+                path_parts = ref_path.split("/")
+                schema_name = path_parts[-1]
+
+                # Parse the schema with forced base model generation
+                # Temporarily override _should_generate_base_model
+                original_method = self._should_generate_base_model
+
+                def force_base_model(*, generates_separate_models: bool = False) -> bool:  # noqa: ARG001
+                    return True
+
+                self._should_generate_base_model = force_base_model  # type: ignore[method-assign]
+                try:
+                    # Re-parse the object to generate the base model
+                    self.parse_obj(schema_name, ref_schema, path_parts)
+                finally:
+                    self._should_generate_base_model = original_method  # type: ignore[method-assign]
+            except Exception:  # noqa: BLE001, S110  # pragma: no cover
+                pass  # Skip if we can't load or parse the schema
+
+    def _parse_raw_impl(self) -> None:  # noqa: PLR0912
+        """Parse OpenAPI raw data implementation."""
         for source, path_parts in self._get_context_source_path_parts():
             if self.validation:
                 warn(

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(