Initial Modular Response initialization

Signed-off-by: Mihai Damaschin <[email protected]>

Author: CostcoFanboy

flask_openapi3/__init__.py

@@ -45,6 +45,6 @@
 from .models.server import Server
 from .models.server import ServerVariable
 from .models.tag import Tag
-from .models.validation_error import UnprocessableEntity
+from .models.validation_error import ErrorResponseModel
 from .openapi import OpenAPI
 from .view import APIView

flask_openapi3/blueprint.py

@@ -113,7 +113,8 @@ def _do_decorator(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
-            method: str = HTTPMethod.GET
+            method: str = HTTPMethod.GET,
+            validation_error_status: str = "422"
     ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]:
         """
         Collects OpenAPI specification information for Flask routes and view functions.
@@ -136,6 +137,8 @@ def _do_decorator(
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Add openapi document UI (swagger, rapidoc and redoc).
                     Defaults to True.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic
+                                    Defaults to 422
         """
         if self.doc_ui is True and doc_ui is True:
             if responses is None:
@@ -178,7 +181,8 @@ def _do_decorator(
                 operation=operation
             )
             # Parse response
-            get_responses(combine_responses, extra_responses, self.components_schemas, operation)
+            get_responses(combine_responses, extra_responses, self.components_schemas, operation,
+                          validation_error_status)
             # Convert route parameter format from /pet/<petId> to /pet/{petId}
             uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}")
             trail_slash = uri.endswith("/")

flask_openapi3/models/validation_error.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, Field
 
 
-class UnprocessableEntity(BaseModel):
+class ErrorResponseModel(BaseModel):
     # More information: https://pydantic-docs.helpmanual.io/usage/models/#error-handling
     loc: Optional[List[str]] = Field(None, title="Location", description="the error's location as a list. ")
     msg: Optional[str] = Field(None, title="Message", description="a computer-readable identifier of the error type.")

flask_openapi3/openapi.py

@@ -45,6 +45,7 @@ def __init__(
             external_docs: Optional[ExternalDocumentation] = None,
             operation_id_callback: Callable = get_operation_id_for_path,
             openapi_extensions: Optional[Dict[str, Any]] = None,
+            validation_error_status: Optional[str] = "422",
             **kwargs: Any
     ) -> None:
         """
@@ -77,6 +78,8 @@ def __init__(
                                    Default to `get_operation_id_for_path` from utils.
             openapi_extensions: Extensions to the OpenAPI Schema.
                                 See https://spec.openapis.org/oas/v3.0.3#specification-extensions.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                                    Defaults to 422
             **kwargs: Additional kwargs to be passed to Flask.
         """
         super(OpenAPI, self).__init__(import_name, **kwargs)
@@ -120,6 +123,9 @@ def __init__(
         # Set OpenAPI extensions
         self.openapi_extensions = openapi_extensions or dict()
 
+        # Set HTTP Response of validation errors within OpenAPI
+        self.validation_error_status = validation_error_status
+
         # Initialize the OpenAPI documentation UI
         if doc_ui:
             self._init_doc()
@@ -294,7 +300,8 @@ def _do_decorator(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
-            method: str = HTTPMethod.GET
+            method: str = HTTPMethod.GET,
+            validation_error_status: str = "422"
     ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]:
         """
         Collects OpenAPI specification information for Flask routes and view functions.
@@ -317,6 +324,8 @@ def _do_decorator(
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Add OpenAPI document UI (swagger, rapidoc, and redoc). Defaults to True.
             method: HTTP method for the operation. Defaults to GET.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic
+                                    Defaults to 422
         """
         if doc_ui is True:
             if responses is None:
@@ -358,7 +367,8 @@ def _do_decorator(
                 operation=operation
             )
             # Parse response
-            get_responses(combine_responses, extra_responses, self.components_schemas, operation)
+            get_responses(combine_responses, extra_responses, self.components_schemas, operation,
+                          validation_error_status)
             # Convert route parameter format from /pet/<petId> to /pet/{petId}
             uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}")
             # Parse method

flask_openapi3/request.py

@@ -93,7 +93,8 @@ def _do_request(
         query: Optional[Type[BaseModel]] = None,
         form: Optional[Type[BaseModel]] = None,
         body: Optional[Type[BaseModel]] = None,
-        path_kwargs: Optional[Dict[Any, Any]] = None
+        validation_error_status: str = "422",
+        path_kwargs: Optional[Dict[Any, Any]] = None,
 ) -> Union[Response, Dict]:
     """
     Validate requests and responses.
@@ -105,6 +106,7 @@ def _do_request(
         query: Query model.
         form: Form model.
         body: Body model.
+        validation_error_status: HTTP Status of the response upon validation failure
         path_kwargs: Path parameters.
 
     Returns:
@@ -135,7 +137,7 @@ def _do_request(
         # Create a JSON response with validation error details
         response = make_response(e.json())
         response.headers["Content-Type"] = "application/json"
-        response.status_code = 422
+        response.status_code = int(validation_error_status)
         return response
 
     return func_kwargs

flask_openapi3/scaffold.py

@@ -55,7 +55,8 @@ def _do_decorator(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
-            method: str = HTTPMethod.GET
+            method: str = HTTPMethod.GET,
+            validation_error_status: str = "422"
     ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]:
         raise NotImplementedError
 
@@ -71,8 +72,9 @@ def create_view_func(
             query,
             form,
             body,
+            validation_error_status="422",
             view_class=None,
-            view_kwargs=None
+            view_kwargs=None,
     ):
         is_coroutine_function = iscoroutinefunction(func)
         if is_coroutine_function:
@@ -85,10 +87,11 @@ async def view_func(**kwargs) -> Response:
                     query=query,
                     form=form,
                     body=body,
+                    validation_error_status=validation_error_status,
                     path_kwargs=kwargs
                 )
                 if isinstance(func_kwargs, Response):
-                    # 422
+                    # validation_error_status from OpenAPI Initialization
                     return func_kwargs
                 # handle async request
                 if view_class:
@@ -112,10 +115,11 @@ def view_func(**kwargs) -> Response:
                     query=query,
                     form=form,
                     body=body,
-                    path_kwargs=kwargs
+                    path_kwargs=kwargs,
+                    validation_error_status=validation_error_status
                 )
                 if isinstance(result, Response):
-                    # 422
+                    # validation_error_status from OpenAPI initialization
                     return result
                 # handle request
                 if view_class:
@@ -153,6 +157,7 @@ def get(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
+            validation_error_status: str = "422",
             **options: Any
     ) -> Callable:
         """
@@ -176,6 +181,8 @@ def get(
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Add openapi document UI (swagger, rapidoc and redoc).
                     Default to True.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                                    Defaults to 422
         """
 
         if extra_form is not None:
@@ -210,10 +217,11 @@ def decorator(func) -> Callable:
                     servers=servers,
                     openapi_extensions=openapi_extensions,
                     doc_ui=doc_ui,
-                    method=HTTPMethod.GET
+                    method=HTTPMethod.GET,
+                    validation_error_status=validation_error_status
                 )
 
-            view_func = self.create_view_func(func, header, cookie, path, query, form, body)
+            view_func = self.create_view_func(func, header, cookie, path, query, form, body, validation_error_status)
             options.update({"methods": [HTTPMethod.GET]})
             self.add_url_rule(rule, view_func=view_func, **options)
 
@@ -239,6 +247,7 @@ def post(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
+            validation_error_status: str = "422",
             **options: Any
     ) -> Callable:
         """
@@ -261,6 +270,8 @@ def post(
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Declares this operation to be shown.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                                    Defaults to 422
         """
         if extra_form is not None:
             warnings.warn(
@@ -294,10 +305,11 @@ def decorator(func) -> Callable:
                     servers=servers,
                     openapi_extensions=openapi_extensions,
                     doc_ui=doc_ui,
-                    method=HTTPMethod.POST
+                    method=HTTPMethod.POST,
+                    validation_error_status=validation_error_status
                 )
 
-            view_func = self.create_view_func(func, header, cookie, path, query, form, body)
+            view_func = self.create_view_func(func, header, cookie, path, query, form, body, validation_error_status)
             options.update({"methods": [HTTPMethod.POST]})
             self.add_url_rule(rule, view_func=view_func, **options)
 
@@ -323,6 +335,7 @@ def put(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
+            validation_error_status: str = "422",
             **options: Any
     ) -> Callable:
         """
@@ -345,6 +358,8 @@ def put(
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Declares this operation to be shown.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                                    Defaults to 422
         """
         if extra_form is not None:
             warnings.warn(
@@ -378,10 +393,11 @@ def decorator(func) -> Callable:
                     servers=servers,
                     openapi_extensions=openapi_extensions,
                     doc_ui=doc_ui,
-                    method=HTTPMethod.PUT
+                    method=HTTPMethod.PUT,
+                    validation_error_status=validation_error_status
                 )
 
-            view_func = self.create_view_func(func, header, cookie, path, query, form, body)
+            view_func = self.create_view_func(func, header, cookie, path, query, form, body, validation_error_status)
             options.update({"methods": [HTTPMethod.PUT]})
             self.add_url_rule(rule, view_func=view_func, **options)
 
@@ -406,6 +422,7 @@ def delete(
             security: Optional[List[Dict[str, List[Any]]]] = None,
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
+            validation_error_status: str = "422",
             doc_ui: bool = True,
             **options: Any
     ) -> Callable:
@@ -429,6 +446,8 @@ def delete(
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Declares this operation to be shown.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                                    Defaults to 422
         """
         if extra_form is not None:
             warnings.warn(
@@ -462,10 +481,11 @@ def decorator(func) -> Callable:
                     servers=servers,
                     openapi_extensions=openapi_extensions,
                     doc_ui=doc_ui,
-                    method=HTTPMethod.DELETE
+                    method=HTTPMethod.DELETE,
+                    validation_error_status=validation_error_status
                 )
 
-            view_func = self.create_view_func(func, header, cookie, path, query, form, body)
+            view_func = self.create_view_func(func, header, cookie, path, query, form, body, validation_error_status)
             options.update({"methods": [HTTPMethod.DELETE]})
             self.add_url_rule(rule, view_func=view_func, **options)
 
@@ -491,6 +511,7 @@ def patch(
             servers: Optional[List[Server]] = None,
             openapi_extensions: Optional[Dict[str, Any]] = None,
             doc_ui: bool = True,
+            validation_error_status: str = "422",
             **options: Any
     ) -> Callable:
         """
@@ -513,6 +534,8 @@ def patch(
             servers: An alternative server array to service this operation.
             openapi_extensions: Allows extensions to the OpenAPI Schema.
             doc_ui: Declares this operation to be shown.
+            validation_error_status: HTTP Status of the response given when a validation error is detected by pydantic.
+                    Defaults to 422
         """
         if extra_form is not None:
             warnings.warn(
@@ -546,10 +569,12 @@ def decorator(func) -> Callable:
                     servers=servers,
                     openapi_extensions=openapi_extensions,
                     doc_ui=doc_ui,
-                    method=HTTPMethod.PATCH
+                    method=HTTPMethod.PATCH,
+                    validation_error_status=validation_error_status
                 )
 
-            view_func = self.create_view_func(func, header, cookie, path, query, form, body)
+            view_func = self.create_view_func(func, header, cookie, path, query, form, body,
+                                              validation_error_status=validation_error_status)
             options.update({"methods": [HTTPMethod.PATCH]})
             self.add_url_rule(rule, view_func=view_func, **options)
 

flask_openapi3/utils.py

@@ -13,7 +13,7 @@
 from .models.common import Schema, MediaType, Encoding, ExtraRequestBody
 from .models.path import Operation, RequestBody, PathItem, Response
 from .models.path import ParameterInType, Parameter
-from .models.validation_error import UnprocessableEntity
+from .models.validation_error import ErrorResponseModel
 
 
 def get_operation(
@@ -283,30 +283,31 @@ def get_responses(
         responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]],
         extra_responses: Dict[str, dict],
         components_schemas: dict,
-        operation: Operation
+        operation: Operation,
+        validation_error_status: str = "422"
 ) -> None:
     if responses is None:
         responses = {}
     _responses = {}
     _schemas = {}
-    if not responses.get("422"):
-        # Handle response 422 for Unprocessable Entity
-        _responses["422"] = Response(
-            description=HTTP_STATUS["422"],
+    if not responses.get(validation_error_status):
+        # Handle response for validation modelling errors
+        _responses[validation_error_status] = Response(
+            description=HTTP_STATUS[validation_error_status],
             content={
                 "application/json": MediaType(
                     **{
                         "schema": Schema(
                             **{
                                 "type": "array",
-                                "items": {"$ref": f"{OPENAPI3_REF_PREFIX}/{UnprocessableEntity.__name__}"}
+                                "items": {"$ref": f"{OPENAPI3_REF_PREFIX}/{ErrorResponseModel.__name__}"}
                             }
                         )
                     }
                 )
             }
         )
-        _schemas[UnprocessableEntity.__name__] = Schema(**UnprocessableEntity.schema())
+        _schemas[ErrorResponseModel.__name__] = Schema(**ErrorResponseModel.schema())
     for key, response in responses.items():
         if response is None:
             # If the response is None, it means HTTP status code "204" (No Content)