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: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = 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

@@ -370,6 +370,8 @@ def _collect_openapi_info(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         method: str = HTTPMethod.GET,
     ) -> ParametersTuple:
@@ -389,6 +391,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.
         """
@@ -437,6 +441,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

@@ -5,14 +5,15 @@
 import json
 from functools import wraps
 from json import JSONDecodeError
-from typing import Any, Type
+from types import UnionType
+from typing import Any, 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 +147,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(

flask_openapi3/scaffold.py

@@ -29,6 +29,8 @@ def _collect_openapi_info(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         method: str = HTTPMethod.GET,
     ) -> ParametersTuple:
@@ -192,6 +194,8 @@ def post(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         **options: Any,
     ) -> Callable:
@@ -211,26 +215,31 @@ 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.
         """
 
         def decorator(func) -> Callable:
-            header, cookie, path, query, form, body, raw = self._collect_openapi_info(
-                rule,
-                func,
-                tags=tags,
-                summary=summary,
-                description=description,
-                external_docs=external_docs,
-                operation_id=operation_id,
-                responses=responses,
-                deprecated=deprecated,
-                security=security,
-                servers=servers,
-                openapi_extensions=openapi_extensions,
-                doc_ui=doc_ui,
-                method=HTTPMethod.POST,
-            )
+            header, cookie, path, query, form, body, raw = \
+                self._collect_openapi_info(
+                    rule,
+                    func,
+                    tags=tags,
+                    summary=summary,
+                    description=description,
+                    external_docs=external_docs,
+                    operation_id=operation_id,
+                    responses=responses,
+                    deprecated=deprecated,
+                    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,
+                )
 
             view_func = self.create_view_func(func, header, cookie, path, query, form, body, raw)
             options.update({"methods": [HTTPMethod.POST]})
@@ -254,6 +263,8 @@ def put(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         **options: Any,
     ) -> Callable:
@@ -273,26 +284,31 @@ 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.
         """
 
         def decorator(func) -> Callable:
-            header, cookie, path, query, form, body, raw = self._collect_openapi_info(
-                rule,
-                func,
-                tags=tags,
-                summary=summary,
-                description=description,
-                external_docs=external_docs,
-                operation_id=operation_id,
-                responses=responses,
-                deprecated=deprecated,
-                security=security,
-                servers=servers,
-                openapi_extensions=openapi_extensions,
-                doc_ui=doc_ui,
-                method=HTTPMethod.PUT,
-            )
+            header, cookie, path, query, form, body, raw = \
+                self._collect_openapi_info(
+                    rule,
+                    func,
+                    tags=tags,
+                    summary=summary,
+                    description=description,
+                    external_docs=external_docs,
+                    operation_id=operation_id,
+                    responses=responses,
+                    deprecated=deprecated,
+                    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,
+                )
 
             view_func = self.create_view_func(func, header, cookie, path, query, form, body, raw)
             options.update({"methods": [HTTPMethod.PUT]})
@@ -316,6 +332,8 @@ def delete(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         **options: Any,
     ) -> Callable:
@@ -335,26 +353,31 @@ 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.
         """
 
         def decorator(func) -> Callable:
-            header, cookie, path, query, form, body, raw = self._collect_openapi_info(
-                rule,
-                func,
-                tags=tags,
-                summary=summary,
-                description=description,
-                external_docs=external_docs,
-                operation_id=operation_id,
-                responses=responses,
-                deprecated=deprecated,
-                security=security,
-                servers=servers,
-                openapi_extensions=openapi_extensions,
-                doc_ui=doc_ui,
-                method=HTTPMethod.DELETE,
-            )
+            header, cookie, path, query, form, body, raw = \
+                self._collect_openapi_info(
+                    rule,
+                    func,
+                    tags=tags,
+                    summary=summary,
+                    description=description,
+                    external_docs=external_docs,
+                    operation_id=operation_id,
+                    responses=responses,
+                    deprecated=deprecated,
+                    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,
+                )
 
             view_func = self.create_view_func(func, header, cookie, path, query, form, body, raw)
             options.update({"methods": [HTTPMethod.DELETE]})
@@ -378,6 +401,8 @@ def patch(
         security: list[dict[str, list[Any]]] | None = None,
         servers: list[Server] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        request_body_description: str | None = None,
+        request_body_required: bool | None = True,
         doc_ui: bool = True,
         **options: Any,
     ) -> Callable:
@@ -397,26 +422,31 @@ 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.
         """
 
         def decorator(func) -> Callable:
-            header, cookie, path, query, form, body, raw = self._collect_openapi_info(
-                rule,
-                func,
-                tags=tags,
-                summary=summary,
-                description=description,
-                external_docs=external_docs,
-                operation_id=operation_id,
-                responses=responses,
-                deprecated=deprecated,
-                security=security,
-                servers=servers,
-                openapi_extensions=openapi_extensions,
-                doc_ui=doc_ui,
-                method=HTTPMethod.PATCH,
-            )
+            header, cookie, path, query, form, body, raw = \
+                self._collect_openapi_info(
+                    rule,
+                    func,
+                    tags=tags,
+                    summary=summary,
+                    description=description,
+                    external_docs=external_docs,
+                    operation_id=operation_id,
+                    responses=responses,
+                    deprecated=deprecated,
+                    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,
+                )
 
             view_func = self.create_view_func(func, header, cookie, path, query, form, body, raw)
             options.update({"methods": [HTTPMethod.PATCH]})