Support multi content type in request body and responses

Author: luolingchun

examples/multi_content_type.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+# @Author  : llc
+# @Time    : 2024/12/27 15:30
+from pydantic import BaseModel
+
+from flask_openapi3 import OpenAPI
+
+app = OpenAPI(__name__)
+
+
+class DogBody(BaseModel):
+    a: int = None
+    b: str = None
+
+    model_config = {
+        "openapi_extra": {
+            "content_type": "application/vnd.dog+json"
+        }
+    }
+
+
+class CatBody(BaseModel):
+    c: int = None
+    d: str = None
+
+    model_config = {
+        "openapi_extra": {
+            "content_type": "application/vnd.cat+json"
+        }
+    }
+
+
+class ContentTypeModel(BaseModel):
+    model_config = {
+        "openapi_extra": {
+            "content_type": "text/csv"
+        }
+    }
+
+
+@app.post("/a", responses={200: DogBody | CatBody | ContentTypeModel})
+def index_a(body: DogBody | CatBody | ContentTypeModel):
+    print(body)
+    return {"hello": "world"}
+
+
+if __name__ == '__main__':
+    app.run(debug=True)

flask_openapi3/blueprint.py

@@ -121,6 +121,8 @@ def _collect_openapi_info(
         security: Optional[list[dict[str, list[Any]]]] = None,
         servers: Optional[list[Server]] = None,
         openapi_extensions: Optional[dict[str, Any]] = None,
+        request_body_description: Optional[str] = None,
+        request_body_required: Optional[bool] = True,
         doc_ui: bool = True,
         method: str = HTTPMethod.GET,
     ) -> ParametersTuple:
@@ -140,6 +142,8 @@ def _collect_openapi_info(
             security: A declaration of which security mechanisms can be used for this operation.
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
+            request_body_description: A brief description of the request body.
+            request_body_required: Determines if the request body is required in the request.
             doc_ui: Declares this operation to be shown. Default to True.
         """
         if self.doc_ui is True and doc_ui is True:
@@ -191,6 +195,12 @@ def _collect_openapi_info(
             parse_method(uri, method, self.paths, operation)
 
             # Parse parameters
-            return parse_parameters(func, components_schemas=self.components_schemas, operation=operation)
+            return parse_parameters(
+                func,
+                components_schemas=self.components_schemas,
+                operation=operation,
+                request_body_description=request_body_description,
+                request_body_required=request_body_required
+            )
         else:
             return parse_parameters(func, doc_ui=False)

flask_openapi3/openapi.py

@@ -375,6 +375,8 @@ def _collect_openapi_info(
         security: Optional[list[dict[str, list[Any]]]] = None,
         servers: Optional[list[Server]] = None,
         openapi_extensions: Optional[dict[str, Any]] = None,
+        request_body_description: Optional[str] = None,
+        request_body_required: Optional[bool] = True,
         doc_ui: bool = True,
         method: str = HTTPMethod.GET,
     ) -> ParametersTuple:
@@ -394,6 +396,8 @@ def _collect_openapi_info(
             security: A declaration of which security mechanisms can be used for this operation.
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
+            request_body_description: A brief description of the request body.
+            request_body_required: Determines if the request body is required in the request.
             doc_ui: Declares this operation to be shown. Default to True.
             method: HTTP method for the operation. Defaults to GET.
         """
@@ -442,6 +446,12 @@ def _collect_openapi_info(
             parse_method(uri, method, self.paths, operation)
 
             # Parse parameters
-            return parse_parameters(func, components_schemas=self.components_schemas, operation=operation)
+            return parse_parameters(
+                func,
+                components_schemas=self.components_schemas,
+                operation=operation,
+                request_body_description=request_body_description,
+                request_body_required=request_body_required
+            )
         else:
             return parse_parameters(func, doc_ui=False)

flask_openapi3/request.py

@@ -3,16 +3,16 @@
 # @Time    : 2022/4/1 16:54
 import inspect
 import json
-from functools import wraps
 from json import JSONDecodeError
-from typing import Any, Optional, Type
+from types import UnionType
+from typing import Any, Optional, Type, Union, get_args, get_origin
 
 from flask import abort, current_app, request
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, RootModel, ValidationError
 from pydantic.fields import FieldInfo
 from werkzeug.datastructures.structures import MultiDict
 
-from .utils import parse_parameters
+from flask_openapi3.utils import is_application_json, parse_parameters
 
 
 def _get_list_value(model: Type[BaseModel], args: MultiDict, model_field_key: str, model_field_value: FieldInfo):
@@ -146,12 +146,20 @@ def _validate_form(form: Type[BaseModel], func_kwargs: dict):
 
 
 def _validate_body(body: Type[BaseModel], func_kwargs: dict):
-    obj = request.get_json(silent=True)
-    if isinstance(obj, str):
-        body_model = body.model_validate_json(json_data=obj)
+    if is_application_json(request.mimetype):
+        if get_origin(body) == UnionType:
+            root_model_list = [model for model in get_args(body)]
+            Body = RootModel[Union[tuple(root_model_list)]]  # type: ignore
+        else:
+            Body = body  # type: ignore
+        obj = request.get_json(silent=True)
+        if isinstance(obj, str):
+            body_model = Body.model_validate_json(json_data=obj)
+        else:
+            body_model = Body.model_validate(obj=obj)
+        func_kwargs["body"] = body_model
     else:
-        body_model = body.model_validate(obj=obj)
-    func_kwargs["body"] = body_model
+        func_kwargs["body"] = request
 
 
 def _validate_request(