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:
@@ -193,6 +197,11 @@ 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

@@ -353,6 +353,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:
@@ -372,6 +374,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.
         """
@@ -423,6 +427,11 @@ 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,13 +3,16 @@
 # @Time    : 2022/4/1 16:54
 import json
 from json import JSONDecodeError
-from typing import Any, Type, Optional, Dict
+from types import UnionType
+from typing import Any, Type, Optional, Dict, get_origin, get_args, Union
 
 from flask import request, current_app, abort
-from pydantic import ValidationError, BaseModel
+from pydantic import ValidationError, BaseModel, RootModel
 from pydantic.fields import FieldInfo
 from werkzeug.datastructures.structures import MultiDict
 
+from flask_openapi3.utils import is_application_json
+
 
 def _get_list_value(model: Type[BaseModel], args: MultiDict, model_field_key: str, model_field_value: FieldInfo):
     if model_field_value.alias and model.model_config.get("populate_by_name"):
@@ -138,12 +141,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[*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(

flask_openapi3/scaffold.py

@@ -32,13 +32,15 @@ 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:
-        raise NotImplementedError   # pragma: no cover
+        raise NotImplementedError  # pragma: no cover
 
     def register_api(self, api) -> None:
-        raise NotImplementedError   # pragma: no cover
+        raise NotImplementedError  # pragma: no cover
 
     def _add_url_rule(
             self,
@@ -48,7 +50,7 @@ def _add_url_rule(
             provide_automatic_options=None,
             **options,
     ) -> None:
-        raise NotImplementedError   # pragma: no cover
+        raise NotImplementedError  # pragma: no cover
 
     @staticmethod
     def create_view_func(
@@ -199,6 +201,8 @@ def post(
             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,
             **options: Any
     ) -> Callable:
@@ -218,6 +222,8 @@ def post(
             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.
         """
 
@@ -236,6 +242,8 @@ def decorator(func) -> Callable:
                     security=security,
                     servers=servers,
                     openapi_extensions=openapi_extensions,
+                    request_body_description=request_body_description,
+                    request_body_required=request_body_required,
                     doc_ui=doc_ui,
                     method=HTTPMethod.POST
                 )
@@ -262,6 +270,8 @@ def put(
             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,
             **options: Any
     ) -> Callable:
@@ -281,6 +291,8 @@ def put(
             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.
         """
 
@@ -299,6 +311,8 @@ def decorator(func) -> Callable:
                     security=security,
                     servers=servers,
                     openapi_extensions=openapi_extensions,
+                    request_body_description=request_body_description,
+                    request_body_required=request_body_required,
                     doc_ui=doc_ui,
                     method=HTTPMethod.PUT
                 )
@@ -325,6 +339,8 @@ def delete(
             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,
             **options: Any
     ) -> Callable:
@@ -344,6 +360,8 @@ def delete(
             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.
         """
 
@@ -362,6 +380,8 @@ def decorator(func) -> Callable:
                     security=security,
                     servers=servers,
                     openapi_extensions=openapi_extensions,
+                    request_body_description=request_body_description,
+                    request_body_required=request_body_required,
                     doc_ui=doc_ui,
                     method=HTTPMethod.DELETE
                 )
@@ -388,6 +408,8 @@ def patch(
             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,
             **options: Any
     ) -> Callable:
@@ -407,6 +429,8 @@ def patch(
             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.
         """
 
@@ -425,6 +449,8 @@ def decorator(func) -> Callable:
                     security=security,
                     servers=servers,
                     openapi_extensions=openapi_extensions,
+                    request_body_description=request_body_description,
+                    request_body_required=request_body_required,
                     doc_ui=doc_ui,
                     method=HTTPMethod.PATCH
                 )

flask_openapi3/types.py

@@ -2,14 +2,16 @@
 # @Author  : llc
 # @Time    : 2023/7/9 15:25
 from http import HTTPStatus
-from typing import Dict, Union, Type, Any, Tuple, Optional
+from typing import Dict, Union, Type, Any, Tuple, Optional, TypeVar
 
 from pydantic import BaseModel
 
 from .models import RawModel
 from .models import SecurityScheme
 
-_ResponseDictValue = Union[Type[BaseModel], Dict[Any, Any], None]
+_MultiBaseModel = TypeVar("_MultiBaseModel", bound=Type[BaseModel])
+
+_ResponseDictValue = Union[Type[BaseModel], _MultiBaseModel, Dict[Any, Any], None]
 
 ResponseDict = Dict[Union[str, int, HTTPStatus], _ResponseDictValue]