How to implement a logging Interceptor for Python async gRPC client?

user_14421537 0
Viewed 433

I'm trying to implement accept / error logging for an asynchronous gRPC clientwith gRPC AsyncIO API. I would like to handle common errors (like StatusCode.UNAVAILABLE) in one place instead of in every request.

It's easy for the synchronous version with response.exception():

class LoggingClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    def __init__(self, logger: Logger):
        self.logger = logger

    def intercept_unary_unary(self, continuation, client_call_details, request):
        self.logger.debug(f"{request=}")
        response = continuation(client_call_details, request)
        if response.exception():
            self.logger.exception(f"{response.code()}")
        return response

But things get more complicated when using an asynchronous interceptor.

I tried to use try / except, expecting await returns a response, but this did not lead to anything, because await of continuation returns an undone UnaryUnaryCall and it has no method .exception:

# this does not work
class LoggingClientInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    def __init__(self, logger: Logger):
        self.logger = logger

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        self.logger.debug(f"{request=}")
        try:
            response = await continuation(client_call_details, request)
            return response
        except Exception as exc:
            self.logger.exception(f"{exc}")

I can await the response code and compare it with OK, and then throw an exception, but it seems to me that this is somehow the wrong way: what if I want to add another interceptor?

code = await response.code()
if code != grpc.StatusCode.OK:
    raise SmthException

I have searched extensively, including the code in the official repository, but have not found good examples for asynchronous interceptors

I will be glad if someone will show me the reference sample.

grpc interceptor asynchronous logging python
1 Answers

Having similar problem and adjust the code a little bit.

class LoggingClientInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    def __init__(self, logger: Logger):
        self.logger = logger

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        self.logger.debug(f"{request=}")
        try:
            undone_call = await continuation(client_call_details, request)
            response = await undone_call
            return response
        except Exception as exc:
            self.logger.exception(f"{exc}")
            raise exc
user_198642860
Related