feat: relaxed request body param handling and improved operationId derivation

Author: Doug Borg

src/openapi_python_generator/language_converters/python/service_generator.py

@@ -1,13 +1,17 @@
 import re
-from typing import Dict
-from typing import List
-from typing import Literal
-from typing import Optional
-from typing import Tuple
-from typing import Union
+from typing import Dict, List, Literal, Optional, Tuple, Union
 
 import click
-from openapi_pydantic.v3.v3_0 import Reference, Schema, Operation, Parameter, RequestBody, Response, MediaType, PathItem
+from openapi_pydantic.v3.v3_0 import (
+    Reference,
+    Schema,
+    Operation,
+    Parameter,
+    RequestBody,
+    Response,
+    MediaType,
+    PathItem,
+)
 
 from openapi_python_generator.language_converters.python import common
 from openapi_python_generator.language_converters.python.common import normalize_symbol
@@ -28,39 +32,52 @@
 
 
 def generate_body_param(operation: Operation) -> Union[str, None]:
+    """Return JSON body expression; tolerant of missing or primitive schema types.
+
+    Behavior:
+    - Reference body -> data.dict()
+    - Array of model-like items -> [i.dict() for i in data]
+    - Array of primitives / unknown -> data
+    - Object / object-like (has properties/allOf/oneOf/anyOf/additionalProperties) -> data
+    - Primitive / missing type -> data
+    """
     if operation.requestBody is None:
         return None
-    else:
-        if isinstance(operation.requestBody, Reference):
-            return "data.dict()"
-
-        if operation.requestBody.content is None:
-            return None  # pragma: no cover
-
-        if operation.requestBody.content.get("application/json") is None:
-            return None  # pragma: no cover
-
-        media_type = operation.requestBody.content.get("application/json")
-
-        if media_type is None:
-            return None  # pragma: no cover
-
-        if isinstance(media_type.media_type_schema, Reference):
-            return "data.dict()"
-        elif isinstance(media_type.media_type_schema, Schema):
-            schema = media_type.media_type_schema
-            if schema.type == "array":
+    if isinstance(operation.requestBody, Reference):
+        return "data.dict()"
+    content = getattr(operation.requestBody, "content", None)
+    if not isinstance(content, dict):
+        return None
+    mt = content.get("application/json")
+    if mt is None:
+        return None
+    schema = getattr(mt, "media_type_schema", None)
+    if isinstance(schema, Reference):
+        return "data.dict()"
+    if isinstance(schema, Schema):
+        # Array handling
+        if getattr(schema, "type", None) == "array":
+            items = getattr(schema, "items", None)
+            if isinstance(items, Reference):
                 return "[i.dict() for i in data]"
-            elif schema.type == "object":
-                return "data"
-            else:
-                raise Exception(
-                    f"Unsupported schema type for request body: {schema.type}"
-                )  # pragma: no cover
-        else:
-            raise Exception(
-                f"Unsupported schema type for request body: {type(media_type.media_type_schema)}"
-            )  # pragma: no cover
+            if isinstance(items, Schema):
+                if getattr(items, "type", None) == "object" or any(
+                    getattr(items, attr, None) for attr in ["properties", "allOf", "oneOf", "anyOf"]
+                ):
+                    return "[i.dict() for i in data]"
+            return "data"
+        # Object-like
+        if (
+            getattr(schema, "type", None) == "object"
+            or any(
+                getattr(schema, attr, None)
+                for attr in ["properties", "allOf", "oneOf", "anyOf", "additionalProperties"]
+            )
+        ):
+            return "data"
+        # Primitive / unspecified
+        return "data"
+    return None
 
 
 def generate_params(operation: Operation) -> str:
@@ -150,17 +167,14 @@ def _generate_params_from_content(content: Union[Reference, Schema]):
     return params + default_params
 
 
-def generate_operation_id(
-    operation: Operation, http_op: str, path_name: Optional[str] = None
-) -> str:
-    if operation.operationId is not None:
+def generate_operation_id(operation: Operation, http_op: str, path_name: Optional[str] = None) -> str:
+    if operation.operationId:
         return common.normalize_symbol(operation.operationId)
-    elif path_name is not None:
-        return common.normalize_symbol(f"{http_op}_{path_name}")
-    else:
-        raise Exception(
-            f"OperationId is not defined for {http_op} of path_name {path_name} --> {operation.summary}"
-        )  # pragma: no cover
+    if path_name:
+        # Insert underscore before parameter placeholders so /lists/{listId} -> lists_{listId}
+        cleaned = re.sub(r"\{([^}]+)\}", r"_\1", path_name)
+        return common.normalize_symbol(f"{http_op}_{cleaned}")
+    raise RuntimeError("Missing operationId and path_name for operation")
 
 
 def _generate_params(
@@ -203,13 +217,10 @@ def generate_return_type(operation: Operation) -> OpReturnType:
     if isinstance(chosen_response, Response) and chosen_response.content is not None:
         media_type_schema = chosen_response.content.get("application/json")
     elif isinstance(chosen_response, Reference):
-        media_type_schema = MediaType(
-            media_type_schema=chosen_response
-        )  # pragma: no cover
+        # Wrap reference as a MediaType so downstream logic can treat uniformly
+        media_type_schema = MediaType(schema=chosen_response)  # type: ignore[arg-type]
     else:
-        return OpReturnType(
-            type=None, status_code=good_responses[0][0], complex_type=False
-        )
+        return OpReturnType(type=None, status_code=good_responses[0][0], complex_type=False)
 
     if isinstance(media_type_schema, MediaType):
         if isinstance(media_type_schema.media_type_schema, Reference):