经过一段时间的研究，这就是我所学到的。
首先，让我们来看看如何用Swagger编辑器制作一个python-flask服务器。
Swagger Editor使用Swagger Editor中编写的定义通过Swagger Codegen生成服务器存根。codegen返回的这个服务器存根使用flask顶部的框架Connexion来处理所有的HTTP请求和响应，包括对swagger定义（swagger.yaml）的验证。
Connexion是一个框架，它使开发python-flask服务器变得很容易，因为它有很多功能，你必须使自己已经内置了，比如参数验证。我们所需要做的就是替换（在本例中是修改）这些connexion验证器。
有三个验证器：

• 参数验证器
• 请求正文验证器
• 响应验证器

默认情况下，它们被Map到flask，但是我们可以在__main__.py文件中轻松地替换它们，我们将看到。
我们的目标是将默认日志和默认错误响应替换为一些自定义的日志和错误响应。我使用了一个自定义的Error模型和一个名为error_response()的函数来准备错误响应，并使用Loguru来记录错误（不是强制性的，您可以保留原始的日志和错误响应）。
为了进行所需的修改，查看connexion验证器代码，我们可以看到它的大部分都可以重用，我们只需要修改：

• 请求正文验证器：__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将它们Map到flask应用（__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的自定义子类，并在__main__.py中的validator_map字典中替换它。
连接文档：https://connexion.readthedocs.io/en/latest/request.html

请原谅我重复第一次发布在https://stackoverflow.com/a/73051652/1630244的答案
您尝试过Connexion before_request功能吗？下面是一个在Connexion验证正文之前记录标题和内容的示例：

import connexion
import logging
from flask import request

logger = logging.getLogger(__name__)
conn_app = connexion.FlaskApp(__name__)

@conn_app.app.before_request
def before_request():
    for h in request.headers:
        logger.debug('header %s', h)
   logger.debug('data %s', request.get_data())