是否可以在 python/flask 服务器存根中的 Swagger 验证之前 运行 自定义代码？
Is it possible to run custom code before Swagger validations in a python/flask server stub?
我正在使用 swagger 编辑器 (OpenApi 2) 在 python 中创建 flask api。当您在 swagger 中定义模型并将其用作请求正文的架构时，swagger 会在 X_controller.py 文件中将控件交给您之前验证正文。
我想在验证发生之前添加一些代码（用于打印日志以进行调试）。 Swagger 只是打印到标准输出错误，如下所示，当您有很多字段时它们没有用（我需要无效的密钥）。
https://host/path validation error: False is not of type 'string'
10.255.0.2 - - [20/May/2020:20:20:20 +0000] "POST /path HTTP/1.1" 400 116 "-" "GuzzleHttp/7"
我知道技术上您可以大摇大摆地删除验证并在您的代码中手动执行它们，但我想继续使用此功能，当它工作时它很棒。
欢迎提出任何关于如何执行此操作的想法或任何能够记录请求的替代方法。
经过一段时间的研究，这就是我学到的东西。
首先让我们看一下使用 Swagger Editor 制作的 python-flask 服务器是如何工作的。
Swagger Editor 使用在 Swagger Editor 中编写的定义通过 Swagger Codegen 生成服务器存根。 codegen 返回的这个服务器存根使用 flask 之上的框架 Connexion 来处理所有 HTTP 请求和响应，包括针对 swagger 定义的验证 (swagger.yaml)。
Connexion 是一个可以轻松开发 python-flask 服务器的框架，因为它有很多您必须自己内置的功能，例如参数验证。我们需要做的就是替换（在本例中为修改）这些连接验证器。
共有三个验证器：
- 参数验证器
- RequestBodyValidator
- 响应验证器
它们默认映射到 flask，但我们可以在 __main__.py 文件中轻松替换它们，正如我们将看到的那样。
我们的目标是将默认日志和默认错误响应替换为一些自定义的。我正在使用一个自定义 Error 模型和一个名为 error_response() 的函数来准备错误响应，并使用 Loguru 来记录错误（不是强制性的，您可以保留原始的）。
要进行所需的更改，查看connexion validators代码，我们可以看到大部分都可以重用，我们只需要修改：
- RequestBodyValidator：
__call__() 和 validate_schema()
- 参数验证器：
__call__()
所以我们只需要创建两个新的class扩展原有的，并复制和修改那些功能。
复制和粘贴时要小心。此代码基于connexion==1.1.15。如果您使用的是不同的版本，您的 class 应该以此为基础。
在新文件中 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:
将它们映射到 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 的自定义子 class 并将其替换到 [=12= 中的 validator_map 字典中].
连接文档：
https://connexion.readthedocs.io/en/latest/request.html
我正在使用 swagger 编辑器 (OpenApi 2) 在 python 中创建 flask api。当您在 swagger 中定义模型并将其用作请求正文的架构时，swagger 会在 X_controller.py 文件中将控件交给您之前验证正文。
我想在验证发生之前添加一些代码（用于打印日志以进行调试）。 Swagger 只是打印到标准输出错误，如下所示，当您有很多字段时它们没有用（我需要无效的密钥）。
https://host/path validation error: False is not of type 'string'
10.255.0.2 - - [20/May/2020:20:20:20 +0000] "POST /path HTTP/1.1" 400 116 "-" "GuzzleHttp/7"
我知道技术上您可以大摇大摆地删除验证并在您的代码中手动执行它们，但我想继续使用此功能，当它工作时它很棒。
欢迎提出任何关于如何执行此操作的想法或任何能够记录请求的替代方法。
经过一段时间的研究，这就是我学到的东西。
首先让我们看一下使用 Swagger Editor 制作的 python-flask 服务器是如何工作的。
Swagger Editor 使用在 Swagger Editor 中编写的定义通过 Swagger Codegen 生成服务器存根。 codegen 返回的这个服务器存根使用 flask 之上的框架 Connexion 来处理所有 HTTP 请求和响应，包括针对 swagger 定义的验证 (swagger.yaml)。
Connexion 是一个可以轻松开发 python-flask 服务器的框架，因为它有很多您必须自己内置的功能，例如参数验证。我们需要做的就是替换（在本例中为修改）这些连接验证器。
共有三个验证器：
- 参数验证器
- RequestBodyValidator
- 响应验证器
它们默认映射到 flask，但我们可以在 __main__.py 文件中轻松替换它们，正如我们将看到的那样。
我们的目标是将默认日志和默认错误响应替换为一些自定义的。我正在使用一个自定义 Error 模型和一个名为 error_response() 的函数来准备错误响应，并使用 Loguru 来记录错误（不是强制性的，您可以保留原始的）。
要进行所需的更改，查看connexion validators代码，我们可以看到大部分都可以重用，我们只需要修改：
- RequestBodyValidator：
__call__()和validate_schema() - 参数验证器：
__call__()
所以我们只需要创建两个新的class扩展原有的，并复制和修改那些功能。
复制和粘贴时要小心。此代码基于connexion==1.1.15。如果您使用的是不同的版本，您的 class 应该以此为基础。
在新文件中 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:
将它们映射到 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 的自定义子 class 并将其替换到 [=12= 中的 validator_map 字典中].
连接文档： https://connexion.readthedocs.io/en/latest/request.html