python - 是否可以在 python/flask 服务器存根中的 Swagger 验证之前运行自定义代码？

经过一段时间的研究，这就是我所学到的。

首先让我们看一下使用 Swagger Editor 制作的 python-flask 服务器是如何工作的。

Swagger Editor 使用 Swagger Editor 中编写的定义通过 Swagger Codegen 生成服务器存根。这个由 codegen 返回的服务器存根使用 Flask 之上的框架 Connexion 来处理所有 HTTP 请求和响应，包括针对 swagger 定义 (swagger.yaml) 的验证。

Connexion 是一个让开发 python-flask 服务器变得容易的框架，因为它有很多你必须让自己内置的功能，比如参数验证。我们需要做的就是替换（在这种情况下修改）这些连接验证器。

有三个验证器：

参数验证器
请求体验证器
响应验证器

它们默认映射到烧瓶，但我们可以在__main__.py文件中轻松替换它们，正如我们将看到的。

我们的目标是将默认日志和默认错误响应替换为一些自定义日志。我正在使用自定义Error模型和error_response()用于准备错误响应的函数，以及用于记录错误的 Loguru（不是强制性的，您可以保留原始错误）。

要进行所需的更改，查看连接验证器代码，我们可以看到大部分都可以重用，我们只需要修改：

RequestBodyValidator：__call__()和validate_schema()
参数验证器：__call__()

所以我们只需要创建两个新的类来扩展原来的类，并复制和修改这些功能。

复制和粘贴时要小心。此代码基于 connexion==1.1.15。如果您使用的是不同的版本，您应该将您的课程基于它。

在一个新文件custom_validators.py中，我们需要：

import json
import functools
from flask import Flask
from loguru import logger
from requests import Response
from jsonschema import ValidationError
from connexion.utils import all_json, is_null
from connexion.exceptions import ExtraParameterProblem
from swagger_server.models import Error
from connexion.decorators.validation import ParameterValidator, RequestBodyValidator

app = Flask(__name__)


def error_response(response: Error) -> Response:
    return app.response_class(
        response=json.dumps(response.to_dict(), default=str),
        status=response.status,
        mimetype='application/json')


class CustomParameterValidator(ParameterValidator):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def __call__(self, function):
        """
        :type function: types.FunctionType
        :rtype: types.FunctionType
        """

        @functools.wraps(function)
        def wrapper(request):

            if self.strict_validation:
                query_errors = self.validate_query_parameter_list(request)
                formdata_errors = self.validate_formdata_parameter_list(request)

                if formdata_errors or query_errors:
                    raise ExtraParameterProblem(formdata_errors, query_errors)

            for param in self.parameters.get('query', []):
                error = self.validate_query_parameter(param, request)
                if error:
                    response = error_response(Error(status=400, description=f'Error: {error}'))
                    return self.api.get_response(response)

            for param in self.parameters.get('path', []):
                error = self.validate_path_parameter(param, request)
                if error:
                    response = error_response(Error(status=400, description=f'Error: {error}'))
                    return self.api.get_response(response)

            for param in self.parameters.get('header', []):
                error = self.validate_header_parameter(param, request)
                if error:
                    response = error_response(Error(status=400, description=f'Error: {error}'))
                    return self.api.get_response(response)

            for param in self.parameters.get('formData', []):
                error = self.validate_formdata_parameter(param, request)
                if error:
                    response = error_response(Error(status=400, description=f'Error: {error}'))
                    return self.api.get_response(response)

            return function(request)

        return wrapper


class CustomRequestBodyValidator(RequestBodyValidator):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def __call__(self, function):
        """
        :type function: types.FunctionType
        :rtype: types.FunctionType
        """

        @functools.wraps(function)
        def wrapper(request):
            if all_json(self.consumes):
                data = request.json

                if data is None and len(request.body) > 0 and not self.is_null_value_valid:
                    # the body has contents that were not parsed as JSON
                    return error_response(Error(
                        status=415,
                        description="Invalid Content-type ({content_type}), JSON data was expected".format(content_type=request.headers.get("Content-Type", ""))
                    ))
                
                error = self.validate_schema(data, request.url)
                if error and not self.has_default:
                    return error

            response = function(request)
            return response

        return wrapper

    def validate_schema(self, data, url):
        if self.is_null_value_valid and is_null(data):
            return None

        try:
            self.validator.validate(data)
        except ValidationError as exception:
            description = f'Validation error. Attribute "{exception.validator_value}" return this error: "{exception.message}"'
            logger.error(description)
            return error_response(Error(
                status=400,
                description=description
            ))

        return None

一旦我们有了验证器，我们必须使用validator_map__main__.py将它们映射到烧瓶应用程序（）：

validator_map = {
    'parameter': CustomParameterValidator,
    'body': CustomRequestBodyValidator,
    'response': ResponseValidator,
}

app = connexion.App(__name__, specification_dir='./swagger/', validator_map=validator_map)
app.app.json_encoder = encoder.JSONEncoder
app.add_api(Path('swagger.yaml'), arguments={'title': 'MyApp'})

如果你还需要替换我在这个例子中没有使用的验证器，只需创建一个 ResponseValidator 的自定义子类，并在 .validator_map 字典中替换它__main__.py。

连接文档： https ://connexion.readthedocs.io/en/latest/request.html

python - 是否可以在 python/flask 服务器存根中的 Swagger 验证之前运行自定义代码？

问题描述

解决方案

推荐阅读