如何优雅地处理 Spyne XSD 异常
How to handle Spyne XSD exceptions gracefully
只要我的 Spyne 应用程序收到请求,就会执行 XSD 验证。这很好,但只要存在 XSD 违规,就会引发错误,我的应用程序 return 会变成 Client.SchemaValidationError,如下所示:
<soap11env:Fault>
<faultcode>soap11env:Client.SchemaValidationError</faultcode>
<faultstring>:25:0:ERROR:SCHEMASV:SCHEMAV_CVC_DATATYPE_VALID_1_2_1: Element '{http://services.sp.pas.ng.org}DateTimeStamp': '2018-07-25T13:01' is not a valid value of the atomic type 'xs:dateTime'.</faultstring>
<faultactor></faultactor>
</soap11env:Fault>
我想知道如何优雅地处理架构验证错误以及 return 我的服务 out_message 的详细信息字段中的详细信息,而不是仅仅提高标准 Client.SchemaValidationError .我想将错误的详细信息存储为变量并将其传递给我的 OperationOne 函数。
这是我的代码,为了敏感起见,我更改了 var 名称。
TNS = "http://services.so.example.org"
class InMessageType(ComplexModel):
__namespace__ = TNS
class Attributes(ComplexModel.Attributes):
declare_order = 'declared'
field_one = Unicode(values=["ONE", "TWO"],
min_occurs=1)
field_two = Unicode(20, min_occurs=1)
field_three = Unicode(20, min_occurs=0)
Confirmation = Unicode(values=["ACCEPTED", "REJECTED"], min_occurs=1)
FileReason = Unicode(200, min_occurs=0)
DateTimeStamp = DateTime(min_occurs=1)
class OperationOneResponse(ComplexModel):
__namespace__ = TNS
class Attributes(ComplexModel.Attributes):
declare_order = 'declared'
ResponseMessage = Unicode(values=["SUCCESS", "FAILURE"], min_occurs=1)
Details = Unicode(min_len=0, max_len=2000)
class ServiceOne(ServiceBase):
@rpc(InMessageType,
_returns=OperationOneResponse,
_out_message_name='OperationOneResponse',
_in_message_name='InMessageType',
_body_style='bare',
)
def OperationOne(ctx, message):
# DO STUFF HERE
# e.g. return {'ResponseMessage': Failure, 'Details': XSDValidationError}
application = Application([ServiceOne],
TNS,
in_protocol=Soap11(validator='lxml'),
out_protocol=Soap11(),
name='ServiceOne',)
wsgi_application = WsgiApplication(application)
if __name__ == '__main__':
pass
我已经考虑过以下方法,但我似乎还不能让它发挥作用:
- 创建子类
MyApplication
并覆盖 call_wrapper()
函数。
- 使用
in_protocol=Soap11(validator=None)
实例化应用程序
- 在调用包装器内部,将协议设置为 Soap11(validator='lxml') 并(以某种方式)调用将验证消息的内容。将其包装在 try/except 块中,如果出现错误,捕获错误并以任何必要的方式处理它。
我只是还没有弄清楚我可以在覆盖的 call_wrapper()
函数中调用什么来实际执行验证。我试过 protocol.decompose_incoming_envelope()
和其他类似的东西,但还没有成功。
覆盖 call_wrapper
将不起作用,因为在调用它之前会引发验证错误。
您应该改用事件子系统。更具体地说,您必须为 method_exception_object
事件注册应用程序级处理程序。
这是一个例子:
def _on_exception_object(ctx):
if isinstance(ctx.out_error, ValidationError):
ctx.out_error = NicerValidationError(...)
app = Application(...)
app.event_manager.add_listener('method_exception_object', _on_exception_object)
查看此测试以获取更多信息:https://github.com/arskom/spyne/blob/4a74cfdbc7db7552bc89c0e5d5c19ed5d0755bc7/spyne/test/test_service.py#L69
根据您的说明,如果您不想回复更好的错误而是回复常规回复,恐怕 Spyne 并不是为了满足该用例而设计的。 "Converting" 将错误的请求处理状态更改为常规状态会使本已繁重的请求处理逻辑不必要地复杂化。
你可以做的是破解响应文档。
一种方法是实现一个额外的 method_exception_document
事件处理程序,其中 <Fault>
标记及其内容可以根据您的喜好进行编辑,甚至可以换出。
不在我的脑海中:
class ValidationErrorReport(ComplexModel):
_type_info = [
('foo', Unicode),
('bar', Integer32),
]
def _on_exception_document(ctx):
fault_elt, = ctx.out_document.xpath("//soap11:Fault", namespaces={'soap11': NS_SOAP11_ENV})
explanation_elt = get_object_as_xml(ValidationErrorReport(...))
fault_parent = fault_elt.parent()
fault_parent.remove(fault_elt)
fault_parent.add(explanation_elt)
以上需要用相关的 Spyne 和 lxml API 仔细检查(也许你可以使用 find()
而不是 xpath()
),但你明白了。
希望对您有所帮助!
只要我的 Spyne 应用程序收到请求,就会执行 XSD 验证。这很好,但只要存在 XSD 违规,就会引发错误,我的应用程序 return 会变成 Client.SchemaValidationError,如下所示:
<soap11env:Fault>
<faultcode>soap11env:Client.SchemaValidationError</faultcode>
<faultstring>:25:0:ERROR:SCHEMASV:SCHEMAV_CVC_DATATYPE_VALID_1_2_1: Element '{http://services.sp.pas.ng.org}DateTimeStamp': '2018-07-25T13:01' is not a valid value of the atomic type 'xs:dateTime'.</faultstring>
<faultactor></faultactor>
</soap11env:Fault>
我想知道如何优雅地处理架构验证错误以及 return 我的服务 out_message 的详细信息字段中的详细信息,而不是仅仅提高标准 Client.SchemaValidationError .我想将错误的详细信息存储为变量并将其传递给我的 OperationOne 函数。
这是我的代码,为了敏感起见,我更改了 var 名称。
TNS = "http://services.so.example.org"
class InMessageType(ComplexModel):
__namespace__ = TNS
class Attributes(ComplexModel.Attributes):
declare_order = 'declared'
field_one = Unicode(values=["ONE", "TWO"],
min_occurs=1)
field_two = Unicode(20, min_occurs=1)
field_three = Unicode(20, min_occurs=0)
Confirmation = Unicode(values=["ACCEPTED", "REJECTED"], min_occurs=1)
FileReason = Unicode(200, min_occurs=0)
DateTimeStamp = DateTime(min_occurs=1)
class OperationOneResponse(ComplexModel):
__namespace__ = TNS
class Attributes(ComplexModel.Attributes):
declare_order = 'declared'
ResponseMessage = Unicode(values=["SUCCESS", "FAILURE"], min_occurs=1)
Details = Unicode(min_len=0, max_len=2000)
class ServiceOne(ServiceBase):
@rpc(InMessageType,
_returns=OperationOneResponse,
_out_message_name='OperationOneResponse',
_in_message_name='InMessageType',
_body_style='bare',
)
def OperationOne(ctx, message):
# DO STUFF HERE
# e.g. return {'ResponseMessage': Failure, 'Details': XSDValidationError}
application = Application([ServiceOne],
TNS,
in_protocol=Soap11(validator='lxml'),
out_protocol=Soap11(),
name='ServiceOne',)
wsgi_application = WsgiApplication(application)
if __name__ == '__main__':
pass
我已经考虑过以下方法,但我似乎还不能让它发挥作用:
- 创建子类
MyApplication
并覆盖call_wrapper()
函数。 - 使用
in_protocol=Soap11(validator=None)
实例化应用程序
- 在调用包装器内部,将协议设置为 Soap11(validator='lxml') 并(以某种方式)调用将验证消息的内容。将其包装在 try/except 块中,如果出现错误,捕获错误并以任何必要的方式处理它。
我只是还没有弄清楚我可以在覆盖的 call_wrapper()
函数中调用什么来实际执行验证。我试过 protocol.decompose_incoming_envelope()
和其他类似的东西,但还没有成功。
覆盖 call_wrapper
将不起作用,因为在调用它之前会引发验证错误。
您应该改用事件子系统。更具体地说,您必须为 method_exception_object
事件注册应用程序级处理程序。
这是一个例子:
def _on_exception_object(ctx):
if isinstance(ctx.out_error, ValidationError):
ctx.out_error = NicerValidationError(...)
app = Application(...)
app.event_manager.add_listener('method_exception_object', _on_exception_object)
查看此测试以获取更多信息:https://github.com/arskom/spyne/blob/4a74cfdbc7db7552bc89c0e5d5c19ed5d0755bc7/spyne/test/test_service.py#L69
根据您的说明,如果您不想回复更好的错误而是回复常规回复,恐怕 Spyne 并不是为了满足该用例而设计的。 "Converting" 将错误的请求处理状态更改为常规状态会使本已繁重的请求处理逻辑不必要地复杂化。
你可以做的是破解响应文档。
一种方法是实现一个额外的 method_exception_document
事件处理程序,其中 <Fault>
标记及其内容可以根据您的喜好进行编辑,甚至可以换出。
不在我的脑海中:
class ValidationErrorReport(ComplexModel):
_type_info = [
('foo', Unicode),
('bar', Integer32),
]
def _on_exception_document(ctx):
fault_elt, = ctx.out_document.xpath("//soap11:Fault", namespaces={'soap11': NS_SOAP11_ENV})
explanation_elt = get_object_as_xml(ValidationErrorReport(...))
fault_parent = fault_elt.parent()
fault_parent.remove(fault_elt)
fault_parent.add(explanation_elt)
以上需要用相关的 Spyne 和 lxml API 仔细检查(也许你可以使用 find()
而不是 xpath()
),但你明白了。
希望对您有所帮助!