| 16 | |
| 17 | |
| 18 | class ErrorInterceptor(grpc.ServerInterceptor): |
| 19 | def intercept_service(self, continuation, handler_call_details): |
| 20 | handler = continuation(handler_call_details) |
| 21 | if handler is None: |
| 22 | return None |
| 23 | |
| 24 | if handler.unary_unary: |
| 25 | return grpc.unary_unary_rpc_method_handler( |
| 26 | lambda req, ctx: exception_wrapper(handler.unary_unary, req, ctx), |
| 27 | request_deserializer=handler.request_deserializer, |
| 28 | response_serializer=handler.response_serializer, |
| 29 | ) |
| 30 | elif handler.unary_stream: |
| 31 | return grpc.unary_stream_rpc_method_handler( |
| 32 | lambda req, ctx: exception_wrapper(handler.unary_stream, req, ctx), |
| 33 | request_deserializer=handler.request_deserializer, |
| 34 | response_serializer=handler.response_serializer, |
| 35 | ) |
| 36 | elif handler.stream_unary: |
| 37 | return grpc.stream_unary_rpc_method_handler( |
| 38 | lambda req, ctx: exception_wrapper(handler.stream_unary, req, ctx), |
| 39 | request_deserializer=handler.request_deserializer, |
| 40 | response_serializer=handler.response_serializer, |
| 41 | ) |
| 42 | elif handler.stream_stream: |
| 43 | return grpc.stream_stream_rpc_method_handler( |
| 44 | lambda req, ctx: exception_wrapper(handler.stream_stream, req, ctx), |
| 45 | request_deserializer=handler.request_deserializer, |
| 46 | response_serializer=handler.response_serializer, |
| 47 | ) |
| 48 | return handler |
no outgoing calls
no test coverage detected