Update service_generator.py

Author: MaxDude132

src/openapi_python_generator/language_converters/python/service_generator.py

@@ -124,7 +124,9 @@ def generate_body_param(operation: Operation) -> Union[str, None]:
     if operation.requestBody is None:
         return None
     else:
-        if isinstance(operation.requestBody, Reference30) or isinstance(operation.requestBody, Reference31):
+        if isinstance(operation.requestBody, Reference30) or isinstance(
+            operation.requestBody, Reference31
+        ):
             return "data.dict()"
 
         if operation.requestBody.content is None:
@@ -138,7 +140,9 @@ def generate_body_param(operation: Operation) -> Union[str, None]:
         if media_type is None:
             return None  # pragma: no cover
 
-        if isinstance(media_type.media_type_schema, (Reference, Reference30, Reference31)):
+        if isinstance(
+            media_type.media_type_schema, (Reference, Reference30, Reference31)
+        ):
             return "data.dict()"
         elif hasattr(media_type.media_type_schema, "ref"):
             # Handle Reference objects from different OpenAPI versions
@@ -150,7 +154,9 @@ def generate_body_param(operation: Operation) -> Union[str, None]:
             elif schema.type == "object":
                 return "data"
             else:
-                raise Exception(f"Unsupported schema type for request body: {schema.type}")  # pragma: no cover
+                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)}"
@@ -183,17 +189,26 @@ def _generate_params_from_content(content: Any):
             required = False
             param_name_cleaned = common.normalize_symbol(param.name)
 
-            if isinstance(param.param_schema, Schema30) or isinstance(param.param_schema, Schema31):
+            if isinstance(param.param_schema, Schema30) or isinstance(
+                param.param_schema, Schema31
+            ):
                 converted_result = (
                     f"{param_name_cleaned} : {type_converter(param.param_schema, param.required).converted_type}"
                     + ("" if param.required else " = None")
                 )
                 required = param.required
-            elif isinstance(param.param_schema, Reference30) or isinstance(param.param_schema, Reference31):
-                converted_result = f"{param_name_cleaned} : {param.param_schema.ref.split('/')[-1]}" + (
-                    ""
-                    if isinstance(param, Reference30) or isinstance(param, Reference31) or param.required
-                    else " = None"
+            elif isinstance(param.param_schema, Reference30) or isinstance(
+                param.param_schema, Reference31
+            ):
+                converted_result = (
+                    f"{param_name_cleaned} : {param.param_schema.ref.split('/')[-1]}"
+                    + (
+                        ""
+                        if isinstance(param, Reference30)
+                        or isinstance(param, Reference31)
+                        or param.required
+                        else " = None"
+                    )
                 )
                 required = isinstance(param, Reference) or param.required
 
@@ -209,11 +224,17 @@ def _generate_params_from_content(content: Any):
         "application/octet-stream",
     ]
 
-    if operation.requestBody is not None and not is_reference_type(operation.requestBody):
+    if operation.requestBody is not None and not is_reference_type(
+        operation.requestBody
+    ):
         # Safe access only if it's a concrete RequestBody object
         rb_content = getattr(operation.requestBody, "content", None)
-        if isinstance(rb_content, dict) and any(rb_content.get(i) is not None for i in operation_request_body_types):
-            get_keyword = [i for i in operation_request_body_types if rb_content.get(i)][0]
+        if isinstance(rb_content, dict) and any(
+            rb_content.get(i) is not None for i in operation_request_body_types
+        ):
+            get_keyword = [
+                i for i in operation_request_body_types if rb_content.get(i)
+            ][0]
             content = rb_content.get(get_keyword)
             if content is not None and hasattr(content, "media_type_schema"):
                 mts = getattr(content, "media_type_schema", None)
@@ -223,7 +244,9 @@ def _generate_params_from_content(content: Any):
                 ):
                     params += f"{_generate_params_from_content(mts)}, "
                 else:  # pragma: no cover
-                    raise Exception(f"Unsupported media type schema for {str(operation)}: {type(mts)}")
+                    raise Exception(
+                        f"Unsupported media type schema for {str(operation)}: {type(mts)}"
+                    )
         # else: silently ignore unsupported body shapes (could extend later)
     # Replace - with _ in params
     params = params.replace("-", "_")
@@ -232,7 +255,9 @@ def _generate_params_from_content(content: Any):
     return params + default_params
 
 
-def generate_operation_id(operation: Operation, http_op: str, path_name: Optional[str] = None) -> str:
+def generate_operation_id(
+    operation: Operation, http_op: str, path_name: Optional[str] = None
+) -> str:
     if operation.operationId is not None:
         return common.normalize_symbol(operation.operationId)
     elif path_name is not None:
@@ -243,7 +268,9 @@ def generate_operation_id(operation: Operation, http_op: str, path_name: Optiona
         )  # pragma: no cover
 
 
-def _generate_params(operation: Operation, param_in: Literal["query", "header"] = "query"):
+def _generate_params(
+    operation: Operation, param_in: Literal["query", "header"] = "query"
+):
     if operation.parameters is None:
         return []
 
@@ -290,7 +317,9 @@ def generate_return_type(operation: Operation) -> OpReturnType:
         media_type_schema = create_media_type_for_reference(chosen_response)
 
     if media_type_schema is None:
-        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 is_media_type(media_type_schema):
         inner_schema = getattr(media_type_schema, "media_type_schema", None)
@@ -307,18 +336,25 @@ def generate_return_type(operation: Operation) -> OpReturnType:
             )
         elif is_schema_type(inner_schema):
             converted_result = type_converter(inner_schema, True)  # type: ignore
-            if "array" in converted_result.original_type and isinstance(converted_result.import_types, list):
+            if "array" in converted_result.original_type and isinstance(
+                converted_result.import_types, list
+            ):
                 matched = re.findall(r"List\[(.+)\]", converted_result.converted_type)
                 if len(matched) > 0:
                     list_type = matched[0]
                 else:  # pragma: no cover
-                    raise Exception(f"Unable to parse list type from {converted_result.converted_type}")
+                    raise Exception(
+                        f"Unable to parse list type from {converted_result.converted_type}"
+                    )
             else:
                 list_type = None
             return OpReturnType(
                 type=converted_result,
                 status_code=good_responses[0][0],
-                complex_type=bool(converted_result.import_types and len(converted_result.import_types) > 0),
+                complex_type=bool(
+                    converted_result.import_types
+                    and len(converted_result.import_types) > 0
+                ),
                 list_type=list_type,
             )
         else:  # pragma: no cover
@@ -348,7 +384,11 @@ def generate_services(
     global _component_params
 
     # Build a lookup dict for component parameters if available
-    if components is not None and hasattr(components, "parameters") and components.parameters is not None:
+    if (
+        components is not None
+        and hasattr(components, "parameters")
+        and components.parameters is not None
+    ):
         _component_params = {}
         for param_name, param_or_ref in components.parameters.items():
             if isinstance(param_or_ref, (Parameter30, Parameter31)):
@@ -358,7 +398,9 @@ def generate_services(
 
     jinja_env = create_jinja_env()
 
-    def generate_service_operation(op: Operation, path_name: str, async_type: bool) -> ServiceOperation:
+    def generate_service_operation(
+        op: Operation, path_name: str, async_type: bool
+    ) -> ServiceOperation:
         # Merge path-level parameters (always required by spec) into the
         # operation-level parameters so they get turned into function args.
         try:
@@ -372,25 +414,36 @@ def generate_service_operation(op: Operation, path_name: str, async_type: bool)
                         if isinstance(p, (Parameter30, Parameter31)):
                             existing_names.add(p.name)
                 for p in path_level_params:
-                    if isinstance(p, (Parameter30, Parameter31)) and p.name not in existing_names:
+                    if (
+                        isinstance(p, (Parameter30, Parameter31))
+                        and p.name not in existing_names
+                    ):
                         if op.parameters is None:
                             op.parameters = []  # type: ignore
                         op.parameters.append(p)  # type: ignore
         except Exception:  # pragma: no cover
-            print(f"Error merging path-level parameters for {path_name}")  # pragma: no cover
+            print(
+                f"Error merging path-level parameters for {path_name}"
+            )  # pragma: no cover
             pass
 
         params = generate_params(op)
         # Fallback: ensure all {placeholders} in path are present as function params
         try:
-            placeholder_names = [m.group(1) for m in re.finditer(r"\{([^}/]+)\}", path_name)]
-            existing_param_names = {p.split(":")[0].strip() for p in params.split(",") if ":" in p}
+            placeholder_names = [
+                m.group(1) for m in re.finditer(r"\{([^}/]+)\}", path_name)
+            ]
+            existing_param_names = {
+                p.split(":")[0].strip() for p in params.split(",") if ":" in p
+            }
             for ph in placeholder_names:
                 norm_ph = common.normalize_symbol(ph)
                 if norm_ph not in existing_param_names and norm_ph:
                     params = f"{norm_ph}: Any, " + params
         except Exception:  # pragma: no cover
-            print(f"Error ensuring path placeholders in params for {path_name}")  # pragma: no cover
+            print(
+                f"Error ensuring path placeholders in params for {path_name}"
+            )  # pragma: no cover
             pass
         operation_id = generate_operation_id(op, http_operation, path_name)
         query_params = generate_query_params(op)
@@ -414,7 +467,9 @@ def generate_service_operation(op: Operation, path_name: str, async_type: bool)
             use_orjson=common.get_use_orjson(),
         )
 
-        so.content = jinja_env.get_template(library_config.template_name).render(**so.model_dump())
+        so.content = jinja_env.get_template(library_config.template_name).render(
+            **so.model_dump()
+        )
 
         if op.tags is not None and len(op.tags) > 0:
             so.tag = normalize_symbol(op.tags[0])
@@ -454,8 +509,16 @@ def generate_service_operation(op: Operation, path_name: str, async_type: bool)
         services.append(
             Service(
                 file_name=f"{tag}_service",
-                operations=[so for so in service_ops if so.tag == tag and not so.async_client],
-                content="\n".join([so.content for so in service_ops if so.tag == tag and not so.async_client]),
+                operations=[
+                    so for so in service_ops if so.tag == tag and not so.async_client
+                ],
+                content="\n".join(
+                    [
+                        so.content
+                        for so in service_ops
+                        if so.tag == tag and not so.async_client
+                    ]
+                ),
                 async_client=False,
                 library_import=library_config.library_name,
                 use_orjson=common.get_use_orjson(),
@@ -466,8 +529,16 @@ def generate_service_operation(op: Operation, path_name: str, async_type: bool)
         services.append(
             Service(
                 file_name=f"async_{tag}_service",
-                operations=[so for so in service_ops if so.tag == tag and so.async_client],
-                content="\n".join([so.content for so in service_ops if so.tag == tag and so.async_client]),
+                operations=[
+                    so for so in service_ops if so.tag == tag and so.async_client
+                ],
+                content="\n".join(
+                    [
+                        so.content
+                        for so in service_ops
+                        if so.tag == tag and so.async_client
+                    ]
+                ),
                 async_client=True,
                 library_import=library_config.library_name,
                 use_orjson=common.get_use_orjson(),