Как определить глобальный обработчик ошибок в gRPC python

Question

Я пытаюсь перехватить любое исключение, которое возникает в любом сервисере, поэтому я могу убедиться, что я распространяю только известные исключения, а не неожиданные, такие как ValueError, TypeError и т. Д.

Я хотел бы иметь возможность отлавливать любые возникшие ошибки и форматировать их или преобразовывать в другие ошибки, чтобы лучше контролировать информацию, которая отображается.

Я не хочу заключать в каждый метод с помощью команды try / кроме.

Я пытался с перехватчиком, но я не могу поймать ошибки там.

Есть ли способ указать обработчик ошибок для сервера grpc? как то, что вы делаете с колбой или любым другим http-сервером?

Answer 1

gRPC Python в настоящее время не поддерживает обработчик глобальных ошибок на стороне сервера. Перехватчик не будет выполнять обработчик сервера внутри intercept_service функция, так что нет возможности попробовать / кроме.

Кроме того, я обнаружил, что реализация перехватчика сервера Python gRPC отличается от того, что они предложили оригинально на 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

Может это вам поможет:)

def _wrap_rpc_behavior(handler, fn):
    if handler is None:
        return None

    if handler.request_streaming and handler.response_streaming:
        behavior_fn = handler.stream_stream
        handler_factory = grpc.stream_stream_rpc_method_handler
    elif handler.request_streaming and not handler.response_streaming:
        behavior_fn = handler.stream_unary
        handler_factory = grpc.stream_unary_rpc_method_handler
    elif not handler.request_streaming and handler.response_streaming:
        behavior_fn = handler.unary_stream
        handler_factory = grpc.unary_stream_rpc_method_handler
    else:
        behavior_fn = handler.unary_unary
        handler_factory = grpc.unary_unary_rpc_method_handler

    return handler_factory(fn(behavior_fn,
                              handler.request_streaming,
                              handler.response_streaming),
                           request_deserializer=handler.request_deserializer,
                           response_serializer=handler.response_serializer)


class TracebackLoggerInterceptor(grpc.ServerInterceptor):

    def intercept_service(self, continuation, handler_call_details):
        def latency_wrapper(behavior, request_streaming, response_streaming):

            def new_behavior(request_or_iterator, servicer_context):
                try:
                    return behavior(request_or_iterator, servicer_context)
                except Exception as err:
                    logger.exception(err, exc_info=True)
            return new_behavior

        return _wrap_rpc_behavior(continuation(handler_call_details),    latency_wrapper)

Answer 3

Как предполагалось в некоторых предыдущих комментариях, я попробовал подход метакласса, который работает довольно хорошо.

Прилагается простой пример, демонстрирующий, как перехватывать вызовы grpc. Вы можете расширить это, предоставив метаклассу список декораторов, которые вы можете применить к каждой функции.

Кроме того, было бы разумно быть более избирательным в отношении методов, к которым вы применяете оболочку. Хорошим вариантом было бы перечислить методы автоматически сгенерированного базового класса и обернуть только их.

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):
   ...