Optimize performance and unit testing

Author: luolingchun

flask_openapi3/blueprint.py

@@ -1,7 +1,6 @@
 # -*- coding: utf-8 -*-
 # @Author  : llc
 # @Time    : 2022/4/1 16:54
-from copy import deepcopy
 from typing import Optional, List, Dict, Any, Callable
 
 from flask import Blueprint
@@ -64,9 +63,8 @@ def __init__(
         self.abp_tags = abp_tags or []
         self.abp_security = abp_security or []
 
-        abp_responses = abp_responses or {}
         # Convert key to string
-        self.abp_responses = convert_responses_key_to_string(abp_responses)
+        self.abp_responses = convert_responses_key_to_string(abp_responses or {})
 
         self.doc_ui = doc_ui
 
@@ -93,7 +91,7 @@ def register_api(self, api: "APIBlueprint") -> None:
             self.paths[uri] = path_item
 
         # Merge component schemas from the nested APIBlueprint
-        self.components_schemas.update(**api.components_schemas)
+        self.components_schemas.update(api.components_schemas)
 
         # Register the nested APIBlueprint as a blueprint
         self.register_blueprint(api)
@@ -145,44 +143,41 @@ def _collect_openapi_info(
             doc_ui: Declares this operation to be shown. Default to True.
         """
         if self.doc_ui is True and doc_ui is True:
-            if responses is None:
-                new_responses = {}
-            else:
-                # Convert key to string
-                new_responses = convert_responses_key_to_string(responses)
+            # Convert key to string
+            new_responses = convert_responses_key_to_string(responses or {})
+
             # Global response: combine API responses
-            combine_responses = deepcopy(self.abp_responses)
-            combine_responses.update(**new_responses)
+            combine_responses = {**self.abp_responses, **new_responses}
+
             # Create operation
             operation = get_operation(
                 func,
                 summary=summary,
                 description=description,
                 openapi_extensions=openapi_extensions
             )
+
             # Set external docs
             operation.externalDocs = external_docs
+
             # Unique string used to identify the operation.
             operation.operationId = operation_id or self.operation_id_callback(
                 name=self.name, path=rule, method=method
             )
+
             # Only set `deprecated` if True, otherwise leave it as None
             operation.deprecated = deprecated
+
             # Add security
-            if security is None:
-                security = []
-            operation.security = security + self.abp_security or None
+            operation.security = (security or []) + self.abp_security or None
+
             # Add servers
             operation.servers = servers
+
             # Store tags
-            tags = tags + self.abp_tags if tags else self.abp_tags
+            tags = (tags or []) + self.abp_tags
             parse_and_store_tags(tags, self.tags, self.tag_names, operation)
-            # Parse parameters
-            header, cookie, path, query, form, body, raw = parse_parameters(
-                func,
-                components_schemas=self.components_schemas,
-                operation=operation
-            )
+
             # Parse response
             get_responses(combine_responses, self.components_schemas, operation)
 
@@ -191,8 +186,8 @@ def _collect_openapi_info(
 
             # Parse method
             parse_method(uri, method, self.paths, operation)
-            return header, cookie, path, query, form, body, raw
-        else:
+
             # Parse parameters
-            header, cookie, path, query, form, body, raw = parse_parameters(func, doc_ui=False)
-            return header, cookie, path, query, form, body, raw
+            return parse_parameters(func, components_schemas=self.components_schemas, operation=operation)
+        else:
+            return parse_parameters(func, doc_ui=False)

flask_openapi3/commands.py

@@ -24,7 +24,7 @@ def openapi_command(output, _format, indent):
         if _format == "yaml":
             try:
                 import yaml  # type: ignore
-            except ImportError:
+            except ImportError:  # pragma: no cover
                 raise ImportError("pyyaml must be installed.")
             openapi = yaml.safe_dump(obj, allow_unicode=True)
         else:

flask_openapi3/models/callback.py

@@ -3,7 +3,7 @@
 # @Time    : 2023/7/4 9:35
 from typing import TYPE_CHECKING, Dict
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no cover
     from .path_item import PathItem
 else:
     PathItem = "PathItem"

flask_openapi3/models/encoding.py

@@ -7,7 +7,7 @@
 
 from .reference import Reference
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no cover
     from .header import Header
 else:
     Header = "Header"

flask_openapi3/models/path_item.py

@@ -10,7 +10,7 @@
 from .reference import Reference
 from .server import Server
 
-if typing.TYPE_CHECKING:
+if typing.TYPE_CHECKING:  # pragma: no cover
     from .operation import Operation
 
 

flask_openapi3/openapi.py

@@ -5,7 +5,6 @@
 import os
 import re
 import sys
-from copy import deepcopy
 from importlib import import_module
 from typing import Optional, List, Dict, Union, Any, Type, Callable
 
@@ -14,7 +13,7 @@
 
 if sys.version_info >= (3, 10):
     from importlib.metadata import entry_points
-else:
+else:  # pragma: no cover
     from importlib_metadata import entry_points  # type: ignore
 
 from .blueprint import APIBlueprint
@@ -108,10 +107,10 @@ def __init__(
         # Set security schemes, responses, paths and components
         self.security_schemes = security_schemes
 
-        responses = responses or {}
         # Convert key to string
-        self.responses = convert_responses_key_to_string(responses)
+        self.responses = convert_responses_key_to_string(responses or {})
 
+        # Initialize instance variables
         self.paths: Dict = dict()
         self.components_schemas: Dict = dict()
         self.components = Components()
@@ -132,7 +131,7 @@ def __init__(
         self.operation_id_callback: Callable = operation_id_callback
 
         # Set OpenAPI extensions
-        self.openapi_extensions = openapi_extensions or dict()
+        self.openapi_extensions = openapi_extensions or {}
 
         # Set HTTP Response of validation errors within OpenAPI
         self.validation_error_status = str(validation_error_status)
@@ -147,7 +146,7 @@ def __init__(
         self.cli.add_command(openapi_command)  # type: ignore
 
         # Initialize specification JSON
-        self.spec_json: Dict = dict()
+        self.spec_json: Dict = {}
 
     def _init_doc(self) -> None:
         """
@@ -174,6 +173,7 @@ def _init_doc(self) -> None:
         )
 
         ui_templates = []
+        # Iterate over all entry points in the "flask_openapi3.plugins" group
         for entry_point in entry_points(group="flask_openapi3.plugins"):
             try:
                 module_path = entry_point.value
@@ -186,7 +186,7 @@ def _init_doc(self) -> None:
                 bp = plugin_register(doc_url=self.doc_url.lstrip("/"))
                 self.register_blueprint(bp, url_prefix=self.doc_prefix)
                 ui_templates.append({"name": plugin_name, "display_name": plugin_display_name})
-            except (ModuleNotFoundError, AttributeError):
+            except (ModuleNotFoundError, AttributeError):  # pragma: no cover
                 import traceback
                 print(f"Warning: plugin '{entry_point.value}' registration failed.")
                 traceback.print_exc()
@@ -281,6 +281,7 @@ def register_api(self, api: APIBlueprint) -> None:
             if tag.name not in self.tag_names:
                 # Append tag to the list of tags
                 self.tags.append(tag)
+
                 # Append tag name to the list of tag names
                 self.tag_names.append(tag.name)
 
@@ -309,6 +310,7 @@ def register_api_view(self, api_view: APIView, view_kwargs: Optional[Dict[Any, A
             if tag.name not in self.tag_names:
                 # Append tag to the list of tags
                 self.tags.append(tag)
+
                 # Append tag name to the list of tag names
                 self.tag_names.append(tag.name)
 
@@ -369,14 +371,12 @@ def _collect_openapi_info(
             method: HTTP method for the operation. Defaults to GET.
         """
         if doc_ui is True:
-            if responses is None:
-                new_responses = {}
-            else:
-                # Convert key to string
-                new_responses = convert_responses_key_to_string(responses)
+            # Convert key to string
+            new_responses = convert_responses_key_to_string(responses or {})
+
             # Global response: combine API responses
-            combine_responses = deepcopy(self.responses)
-            combine_responses.update(**new_responses)
+            combine_responses = {**self.responses, **new_responses}
+
             # Create operation
             operation = get_operation(
                 func,
@@ -386,34 +386,34 @@ def _collect_openapi_info(
             )
             # Set external docs
             operation.externalDocs = external_docs
+
             # Unique string used to identify the operation.
             operation.operationId = operation_id or self.operation_id_callback(
                 name=func.__name__, path=rule, method=method
             )
+
             # Only set `deprecated` if True, otherwise leave it as None
             operation.deprecated = deprecated
+
             # Add security
             operation.security = security
+
             # Add servers
             operation.servers = servers
+
             # Store tags
-            if tags is None:
-                tags = []
-            parse_and_store_tags(tags, self.tags, self.tag_names, operation)
-            # Parse parameters
-            header, cookie, path, query, form, body, raw = parse_parameters(
-                func,
-                components_schemas=self.components_schemas,
-                operation=operation
-            )
+            parse_and_store_tags(tags or [], self.tags, self.tag_names, operation)
+
             # Parse response
             get_responses(combine_responses, self.components_schemas, operation)
+
             # Convert a route parameter format from /pet/<petId> to /pet/{petId}
             uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}")
+
             # Parse method
             parse_method(uri, method, self.paths, operation)
-            return header, cookie, path, query, form, body, raw
-        else:
+
             # Parse parameters
-            header, cookie, path, query, form, body, raw = parse_parameters(func, doc_ui=False)
-            return header, cookie, path, query, form, body, raw
+            return parse_parameters(func, components_schemas=self.components_schemas, operation=operation)
+        else:
+            return parse_parameters(func, doc_ui=False)

flask_openapi3/request.py

@@ -17,30 +17,33 @@ def _validate_header(header: Type[BaseModel], func_kwargs):
         key_title = key.replace("_", "-").title()
         # Add original key
         if key_title in request_headers.keys():
-            request_headers[value.alias or key] = request_headers[key_title]
-    func_kwargs.update({"header": header.model_validate(obj=request_headers)})
+            if value.alias:
+                request_headers[value.alias] = request_headers[key] = request_headers[key_title]
+            else:
+                request_headers[key] = request_headers[key_title]
+    func_kwargs["header"] = header.model_validate(obj=request_headers)
 
 
 def _validate_cookie(cookie: Type[BaseModel], func_kwargs):
     request_cookies = dict(request.cookies)
-    func_kwargs.update({"cookie": cookie.model_validate(obj=request_cookies)})
+    func_kwargs["cookie"] = cookie.model_validate(obj=request_cookies)
 
 
 def _validate_path(path: Type[BaseModel], path_kwargs, func_kwargs):
-    func_kwargs.update({"path": path.model_validate(obj=path_kwargs)})
+    func_kwargs["path"] = path.model_validate(obj=path_kwargs)
 
 
 def _validate_query(query: Type[BaseModel], func_kwargs):
     request_args = request.args
     query_dict = {}
     for k, v in query.model_fields.items():
         if get_origin(v.annotation) is list:
-            value = request_args.getlist(v.alias or k)
+            value = request_args.getlist(v.alias or k) or request_args.getlist(k)
         else:
-            value = request_args.get(v.alias or k)  # type:ignore
+            value = request_args.get(v.alias or k) or request_args.get(k)  # type:ignore
         if value is not None:
             query_dict[k] = value
-    func_kwargs.update({"query": query.model_validate(obj=query_dict)})
+    func_kwargs["query"] = query.model_validate(obj=query_dict)
 
 
 def _validate_form(form: Type[BaseModel], func_kwargs):
@@ -50,25 +53,25 @@ def _validate_form(form: Type[BaseModel], func_kwargs):
     for k, v in form.model_fields.items():
         if get_origin(v.annotation) is list:
             if get_args(v.annotation)[0] is FileStorage:
-                value = request_files.getlist(v.alias or k)
+                value = request_files.getlist(v.alias or k) or request_files.getlist(k)
             else:
                 value = []
-                for i in request_form.getlist(v.alias or k):
+                for i in request_form.getlist(v.alias or k) or request_form.getlist(k):
                     try:
                         value.append(json.loads(i))
                     except (JSONDecodeError, TypeError):
                         value.append(i)  # type:ignore
         elif v.annotation is FileStorage:
-            value = request_files.get(v.alias or k)  # type:ignore
+            value = request_files.get(v.alias or k) or request_files.get(k)  # type:ignore
         else:
-            _value = request_form.get(v.alias or k)
+            _value = request_form.get(v.alias or k) or request_form.get(k)
             try:
                 value = json.loads(_value)  # type:ignore
             except (JSONDecodeError, TypeError):
                 value = _value  # type:ignore
         if value is not None:
             form_dict[k] = value
-    func_kwargs.update({"form": form.model_validate(obj=form_dict)})
+    func_kwargs["form"] = form.model_validate(obj=form_dict)
 
 
 def _validate_body(body: Type[BaseModel], func_kwargs):
@@ -77,7 +80,7 @@ def _validate_body(body: Type[BaseModel], func_kwargs):
         body_model = body.model_validate_json(json_data=obj)
     else:
         body_model = body.model_validate(obj=obj)
-    func_kwargs.update({"body": body_model})
+    func_kwargs["body"] = body_model
 
 
 def _validate_request(
@@ -103,14 +106,14 @@ def _validate_request(
         path_kwargs: Path parameters.
 
     Returns:
-        Union[Response, Dict]: Request kwargs.
+        Dict: Request kwargs.
 
     Raises:
         ValidationError: If validation fails.
     """
 
     # Dictionary to store func kwargs
-    func_kwargs: Dict = dict()
+    func_kwargs: Dict = {}
 
     try:
         # Validate header, cookie, path, and query parameters
@@ -127,7 +130,7 @@ def _validate_request(
         if body:
             _validate_body(body, func_kwargs)
         if raw:
-            func_kwargs.update({"raw": request})
+            func_kwargs["raw"] = request
     except ValidationError as e:
         # Create a response with validation error details
         validation_error_callback = getattr(current_app, "validation_error_callback")