如何在grpcpython中定义全局错误处理程序

Question

我试图捕捉任何服务程序中引发的异常，这样我就可以确保只传播已知的异常，而不是像ValueError、TypeError等意外的异常

我希望能够捕捉到任何引发的错误，并将其格式化或转换为其他错误，以便更好地控制所暴露的信息。在

我不想用try/except将每个服务程序方法括起来。在

我试过用拦截器，但我没能捕捉到那里的错误。在

有没有办法为grpc服务器指定一个错误处理程序？就像你对flask或其他http服务器所做的一样？在

Answer 1

grpcpython目前不支持服务器端全局错误处理程序。拦截器不会在intercept_service函数内执行服务器处理程序，因此无法尝试/except。在

另外，我发现grpcpython服务器拦截器实现与他们最初在L13-Python-Interceptors.md#server-interceptors上提出的实现不同。如果实现坚持最初的设计，我们可以使用拦截器作为全局错误处理程序，并使用handler和request/request_iterator。在

# Current Implementation
intercept_service(self, continuation, handler_call_details)

# Original Design
intercept_unary_unary_handler(self, handler, method, request, servicer_context)
intercept_unary_stream_handler(self, handler, method, request, servicer_context)
intercept_stream_unary_handler(self, handler, method, request_iterator, servicer_context)
intercept_stream_stream_handler(self, handler, method, request_iterator, servicer_context)

请向https://github.com/grpc/grpc/issues提交功能请求问题。在

Answer 2

正如前面的一些评论所建议的，我尝试了元类方法，它非常有效。在

附件是一个简单的例子来演示如何拦截grpc调用。 您可以通过向元类提供一个可以应用于每个函数的decorator列表来扩展它。在

另外，对于应用包装器的方法，最好是更具选择性。一个好的选择是列出自动生成的基类的方法，并且只包装这些方法。在

from types import FunctionType
from functools import wraps


def wrapper(method):
    @wraps(method)
    def wrapped(*args, **kwargs):
        # do stuff here
        return method(*args, **kwargs)

    return wrapped


class ServicerMiddlewareClass(type):
    def __new__(meta, classname, bases, class_dict):
        new_class_dict = {}

        for attribute_name, attribute in class_dict.items():
            if isinstance(attribute, FunctionType):
                # replace it with a wrapped version
                attribute = wrapper(attribute)

            new_class_dict[attribute_name] = attribute

        return type.__new__(meta, classname, bases, new_class_dict)


# In order to use
class MyGrpcService(grpc.MyGrpcServicer, metaclass=ServicerMiddlewareClass):
   ...