Controlling context manager in a meta class

I would like to know if it's possible to control the context automatically in a metaclass and decorator. I have written a decorator function that creates the stub from the grpc insecure channel:

def grpc_factory(grpc_server_address: str):
    print("grpc_factory")
    def grpc_connect(func):
        print("grpc_connect")
        def grpc_connect_wrapper(*args, **kwargs):
            with grpc.insecure_channel(grpc_server_address) as channel:
                stub = AnalyserStub(channel)
                return func(*args, stub=stub, **kwargs)
        return grpc_connect_wrapper
    return grpc_connect

I have then created a metaclass that uses the context manager with every method that starts with grpc_ and then injects the stub into the methods kwargs:

class Client(type):
    @classmethod
    def __prepare__(metacls, name, bases, **kwargs):
        return super().__prepare__(name, bases, **kwargs)

    def __new__(cls, name, bases, attrs, **kwargs):
        if "grpc_server_address" not in kwargs:
            raise ValueError("""grpc_server_address is required on client class, see below example\n
            class MyClient(AnalyserClient, metaclass=Client, grpc_server_address='localhost:50051')""")
        for key, value in attrs.items():
            if callable(value) and key.startswith("grpc_"):
                attrs[key] = grpc_factory(kwargs["grpc_server_address"])(value)
        return super().__new__(cls, name, bases, attrs)

From this, I'd like to create all of the methods from the proto file not implemented errors:

class AnalyserClient(metaclass=Client, grpc_server_address="localhost:50051"):
    def grpc_analyse(self, *args, **kwargs):
        raise NotImplementedError("grpc_analyse is not implemented")

With a final use case of the class below with the stub placed into the methods args:

class AnalyserClient(AC, metaclass=Client, grpc_server_address="localhost:50051"):
    def grpc_analyse(self, text, stub) -> str:
        print("Analysing text: {}".format(text))
        print("Stub is ", stub)
        stub.AnalyseSentiment(text)
        return "Analysed"

I am getting this error which I assume means the channel is no longer open but I'm not sure how this could be done better to ensure all users have a simple interface with safety around using the services defined in the proto file.

grpc_factory
grpc_connect
grpc_factory
grpc_connect
Inside grpc_connect_wrapper
Created channel
Analysing text: Hello World
Stub is  <grpc_implementation.protos.analyse_pb2_grpc.AnalyserStub object at 0x7f29d7726670>
ERROR:grpc._common:Exception serializing message!
Traceback (most recent call last):
  File "/python/venv/lib/python3.8/site-packages/grpc/_common.py", line 86, in _transform
    return transformer(message)
TypeError: descriptor 'SerializeToString' for 'google.protobuf.pyext._message.CMessage' objects doesn't apply to a 'str' object
Traceback (most recent call last):
  File "run_client.py", line 27, in <module>
    client.grpc_analyse("Hello World")
  File "/python/grpc_implementation/client/client.py", line 15, in grpc_connect_wrapper
    return func(*args, stub=stub, **kwargs)
  File "run_client.py", line 11, in grpc_analyse
    stub.AnalyseSentiment(text)
  File "/python/venv/lib/python3.8/site-packages/grpc/_channel.py", line 944, in __call__
    state, call, = self._blocking(request, timeout, metadata, credentials,
  File "/python/venv/lib/python3.8/site-packages/grpc/_channel.py", line 924, in _blocking
    raise rendezvous  # pylint: disable-msg=raising-bad-type
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.INTERNAL
        details = "Exception serializing request!"
        debug_error_string = "None"

The proto file is:

syntax = "proto3";

package analyse;
option go_package = "./grpc_implementation";

service Analyser {
  rpc AnalyseSentiment(SentimentRequest) returns (SentimentResponse) {}
}

message SentimentRequest {
  string text = 1;
}

message SentimentResponse {
  string sentiment = 1;
}

Below is the class I am trying to emulate after the metaclass has added decorator.

class AnalyserClientTrad:
    def __init__(self, host: str = "localhost:50051"):
        self.host = host

    def grpc_analyse(self, text: str):
        with grpc.insecure_channel(self.host) as channel:
            stub = AnalyserStub(channel)
            response = stub.AnalyseSentiment(SentimentRequest(text=text))
            return response.sentiment

client = AnalyserClientTrad()
print(client.grpc_analyse("Hello, world!"))

I have further tested this through adding the decorator traditionally which also works:

def grpc_factory(grpc_server_address: str):
    def grpc_connect(func):
        def grpc_connect_wrapper(*args, **kwargs):
            with grpc.insecure_channel(grpc_server_address) as channel:
                stub = AnalyserStub(channel)
                return func(*args, stub=stub, **kwargs)
        return grpc_connect_wrapper
    return grpc_connect



class AnalyserClientTradWithDecs:
    @grpc_factory("localhost:50051")
    def grpc_analyse(self, text: str, stub: AnalyserStub):
        response = stub.AnalyseSentiment(SentimentRequest(text=text))
        return response.sentiment

def run_client_with_decorator():
    client = AnalyserClientTradWithDecs()
    print(client.grpc_analyse("Hello, world!"))

Any help would be appreciated.

Solution

You have a problem in this part of the code, that your are not setting an expected proto object instead you are setting string

class AnalyserClient(AC, metaclass=Client, grpc_server_address="localhost:50051"):
    def grpc_analyse(self, text, stub) -> str:
        print("Analysing text: {}".format(text))
        print("Stub is ", stub)
        stub.AnalyseSentiment(text) #--> Error, use a proto object here.
        return "Analysed"

The correct way would be to change the line stub.AnalyseSentiment(text) , with

stub.AnalyseSentiment(SentimentRequest(text=text))