WIP: OpenAPI 3.1.0 Documentation Generation

Author: smt5541

flask_parameter_validation/docs_blueprint.py

@@ -1,7 +1,12 @@
+import json
+import warnings
+from typing import Optional
+
 import flask
 from flask import Blueprint, current_app, jsonify
-
 from flask_parameter_validation import ValidateParameters
+from flask_parameter_validation.exceptions.exceptions import ConfigurationError
+import re
 
 docs_blueprint = Blueprint(
     "docs", __name__, url_prefix="/docs", template_folder="./templates"
@@ -38,6 +43,8 @@ def get_function_docs(func):
                 "docstring": format_docstring(fdocs.get("docstring")),
                 "decorators": fdocs.get("decorators"),
                 "args": extract_argument_details(fdocs),
+                "deprecated": fdocs.get("deprecated"),
+                "responses": fdocs.get("openapi_responses"),
             }
     return None
 
@@ -141,3 +148,188 @@ def docs_json():
             "default_theme": config.get("FPV_DOCS_DEFAULT_THEME", "light"),
         }
     )
+
+
+def fpv_error(message):
+    return jsonify({"error": message})
+
+
+def parameter_required(param):
+    if param["type"].startswith("Optional["):
+        return False
+    elif "default" in param["loc_args"]:
+        return False
+    return True
+
+def generate_json_schema_helper(param, param_type, parent_group=None):
+    match = re.match(r'(\w+)\[([\w\[\] ,.]+)]', param_type)
+    if match:
+        type_group = match.group(1)
+        type_params = match.group(2)
+        return generate_json_schema_helper(param, type_params, parent_group=type_group)
+    elif "|" in param_type and "[" not in param_type:  # Handle Union shorthand as Union
+        return generate_json_schema_helper(param, f"Union[{param_type.replace('|', ',')}]", parent_group=parent_group)
+    else:
+        schemas = []
+        param_types = [param_type]
+        if parent_group in ["Union", "Optional"]:
+            if "," in param_type:
+                param_types = [p.strip() for p in param_type.split(",")]
+        for p in param_types:
+            print(f"{param['name']}: {p}")
+            subschema = {}
+            if p == "str":
+                subschema["type"] = "string"
+                if "min_str_length" in param["loc_args"]:
+                    subschema["minLength"] = param["loc_args"]["min_str_length"]
+                if "max_str_length" in param["loc_args"]:
+                    subschema["maxLength"] = param["loc_args"]["max_str_length"]
+                # TODO: Is it possible to make this work with whitelist, blacklist and pattern simultaneously?
+            elif p == "int":
+                subschema["type"] = "integer"
+                if "min_int" in param["loc_args"]:
+                    subschema["minimum"] = param["loc_args"]["min_int"]
+                if "max_int" in param["loc_args"]:
+                    subschema["maximum"] = param["loc_args"]["max_int"]
+            elif p == "bool":
+                subschema["type"] = "boolean"
+            elif p == "float":
+                subschema["type"] = "number"
+            elif p in ["datetime", "datetime.datetime"]:
+                subschema["type"] = "string"
+                subschema["format"] = "date-time"
+                if "datetime_format" in param["loc_args"]:
+                    warnings.warn("datetime_format cannot be translated to JSON Schema, please use ISO8601 date-time",
+                                  Warning, stacklevel=2)
+            elif p in ["date", "datetime.date"]:
+                subschema["type"] = "string"
+                subschema["format"] = "date"
+            elif p in ["time", "datetime.time"]:
+                subschema["type"] = "string"
+                subschema["format"] = "time"
+            elif p == "dict":
+                subschema["type"] = "object"
+            elif p in ["None", "NoneType"]:
+                subschema["type"] = "null"
+            else:
+                print(f"Unexpected type: {p}")
+            schemas.append(subschema)
+        if len(schemas) == 1 and parent_group is None:
+            return schemas[0]
+        elif parent_group in ["Optional", "Union"]:
+            return {"oneOf": schemas}
+        elif parent_group == "List":
+            schema = {"type": "array", "items": schemas[0]}
+            if "min_list_length" in param["loc_args"]:
+                schema["minItems"] = param["loc_args"]["min_list_length"]
+            if "max_list_length" in param["loc_args"]:
+                schema["maxItems"] = param["loc_args"]["max_list_length"]
+            return schema
+        else:
+            print(f"Unexpected situation: {param_type}, {parent_group}")
+
+
+def generate_json_schema_for_parameter(param):
+    return generate_json_schema_helper(param, param["type"])
+
+
+def generate_json_schema_for_parameters(params):
+    schema = {
+        "type": "object",
+        "properties": {},
+        "required": []
+    }
+    for p in params:
+        schema_parameter_name = p["name"] if "alias" not in p["loc_args"] else p["loc_args"]["alias"]
+        if "json_schema" in p["loc_args"]:
+            schema["properties"][schema_parameter_name] = p["loc_args"]["json_schema"]
+        else:
+            schema["properties"][schema_parameter_name] = generate_json_schema_for_parameter(p)
+        if parameter_required(p):
+            schema["required"].append(schema_parameter_name)
+    return schema
+
+def generate_openapi_paths_object():
+    oapi_paths = {}
+    for route in get_route_docs():
+        oapi_path_route = re.sub(r'<(\w+):(\w+)>', r'{\2}', route['rule'])
+        oapi_path_route = re.sub(r'<(\w+)>', r'{\1}', oapi_path_route)
+        print(f"Adding {route['rule']} to paths as {oapi_path_route}")
+        oapi_path_item = {}
+        oapi_operation = {}  # tags, summary, description, externalDocs, operationId, parameters, requestBody, responses, callbacks, deprecated, security, servers
+        oapi_parameters = []
+        oapi_request_body = {"content": {}}
+        for arg_loc in route["args"]:
+            if arg_loc == "Form":
+                oapi_request_body["content"]["application/x-www-form-urlencoded"] = {
+                    "schema": generate_json_schema_for_parameters(route["args"][arg_loc])}
+            elif arg_loc == "Json":
+                oapi_request_body["content"]["application/json"] = {
+                    "schema": generate_json_schema_for_parameters(route["args"][arg_loc])}
+            elif arg_loc == "File":  # See https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#considerations-for-file-uploads
+                for arg in route["args"][arg_loc]:
+                    if "content_types" in arg["loc_args"]:
+                        for content_type in arg["loc_args"]["content_types"]:
+                            oapi_request_body["content"][content_type] = {}
+                    else:
+                        oapi_request_body["content"]["application/octet-stream"] = {}
+            elif arg_loc in ["Route", "Query"]:
+                for arg in route["args"][arg_loc]:
+                    if "alias" in arg["loc_args"]:
+                        oapi_path_route = oapi_path_route.replace(f'{{{arg["name"]}}}',
+                                                                  f'{{{arg["loc_args"]["alias"]}}}')
+                    schema_arg_name = arg["name"] if "alias" not in arg["loc_args"] else arg["loc_args"]["alias"]
+                    if arg_loc == "Query" or (arg_loc == "Route" and f"{{{schema_arg_name}}}" in oapi_path_route):
+                        parameter = {
+                            "name": schema_arg_name,
+                            "in": "path" if arg_loc == "Route" else "query",
+                            "required": True if arg_loc == "Route" else parameter_required(arg),
+                            "schema": arg["loc_args"]["json_schema"] if "json_schema" in arg[
+                                "loc_args"] else generate_json_schema_for_parameter(arg),
+                        }
+                        if "deprecated" in arg["loc_args"] and arg["loc_args"]["deprecated"]:
+                            parameter["deprecated"] = arg["loc_args"]["deprecated"]
+                        oapi_parameters.append(parameter)
+        if len(oapi_parameters) > 0:
+            oapi_operation["parameters"] = oapi_parameters
+        if len(oapi_request_body["content"].keys()) > 0:
+            oapi_operation["requestBody"] = oapi_request_body
+        print(route["decorators"])
+        for decorator in route["decorators"]:
+            for partial_decorator in ["@warnings.deprecated", "@deprecated"]:  # Support for PEP 702 in Python 3.13
+                if partial_decorator in decorator:
+                    oapi_operation["deprecated"] = True
+        if route["deprecated"]:  # Fallback on kwarg passed to @ValidateParameters()
+            oapi_operation["deprecated"] = route["deprecated"]
+        if route["responses"]:
+            oapi_operation["responses"] = route["responses"]
+        for method in route["methods"]:
+            if method not in ["OPTIONS", "HEAD"]:
+                oapi_path_item[method.lower()] = oapi_operation
+        if oapi_path_route in oapi_paths:
+            oapi_paths[oapi_path_route] = oapi_paths[oapi_path_route] | oapi_path_item
+        else:
+            oapi_paths[oapi_path_route] = oapi_path_item
+    return oapi_paths
+
+
+
+@docs_blueprint.route("/openapi")
+def docs_openapi():
+    """
+    Provide the documentation in OpenAPI format
+    """
+    config = flask.current_app.config
+    if not config.get("FPV_OPENAPI_ENABLE", False):
+        return fpv_error("FPV_OPENAPI_ENABLE is not set, and defaults to False")
+
+    supported_versions = ["3.1.0"]
+    openapi_base = config.get("FPV_OPENAPI_BASE", {"openapi": None})
+    if openapi_base["openapi"] not in supported_versions:
+        return fpv_error(f"Flask-Parameter-Validation only supports OpenAPI {', '.join(supported_versions)}, {openapi_base['openapi']} provided")
+    if "paths" in openapi_base:
+        return fpv_error(f"Flask-Parameter-Validation will overwrite the paths value of FPV_OPENAPI_BASE")
+    openapi_paths = generate_openapi_paths_object()
+    openapi_document = json.loads(json.dumps(openapi_base))
+    openapi_document["paths"] = openapi_paths
+    return jsonify(openapi_document)

flask_parameter_validation/exceptions/exceptions.py

@@ -24,5 +24,15 @@ def __init__(self, error_string, input_name, input_type):
         )
         super().__init__(error_string, input_name, input_type)
 
+    def __str__(self):
+        return self.message
+
+class ConfigurationError(Exception):
+    """Called if app configuration is invalid"""
+
+    def __init__(self, message):
+        self.message = message
+        super().__init__(message)
+
     def __str__(self):
         return self.message

flask_parameter_validation/parameter_types/parameter.py

@@ -89,6 +89,7 @@ def validate(self, value):
                 except JSONSchemaValidationError as e:
                     raise ValueError(f"failed JSON Schema validation: {e.args[0]}")
         elif type(value) is dict:
+            # TODO: Make json_schema work for all parameters besides FileStorage and datetime.*? Or maybe even datetime.*?
             if self.json_schema is not None:
                 try:
                     jsonschema.validate(value, self.json_schema)

flask_parameter_validation/parameter_types/query.py

@@ -10,7 +10,8 @@
 class Query(Parameter):
     name = "query"
 
-    def __init__(self, default=None, **kwargs):
+    def __init__(self, default=None, deprecated=False, **kwargs):
+        self.deprecated = deprecated
         super().__init__(default, **kwargs)
 
     def convert(self, value, allowed_types):

flask_parameter_validation/parameter_types/route.py

@@ -8,7 +8,8 @@
 class Route(Parameter):
     name = "route"
 
-    def __init__(self, default=None, **kwargs):
+    def __init__(self, default=None, deprecated=False, **kwargs):
+        self.deprecated = deprecated
         super().__init__(default, **kwargs)
 
     def convert(self, value, allowed_types):

flask_parameter_validation/parameter_validation.py

@@ -18,8 +18,10 @@ class ValidateParameters:
     def get_fn_list(cls):
         return fn_list
 
-    def __init__(self, error_handler=None):
+    def __init__(self, error_handler=None, route_deprecated=False, openapi_responses=None):
         self.custom_error_handler = error_handler
+        self.route_deprecated = route_deprecated
+        self.openapi_responses = openapi_responses
 
     def __call__(self, f):
         """
@@ -37,6 +39,8 @@ def __call__(self, f):
             "argspec": argspec,
             "docstring": f.__doc__.strip() if f.__doc__ else None,
             "decorators": decorators.copy(),
+            "deprecated": self.route_deprecated,
+            "openapi_responses": self.openapi_responses,
         }
         fn_list[fsig] = fdocs