在元类中控制上下文管理器

Question

我想知道是否可以在元级和装饰器中自动控制上下文。 该功能从GRPC不安全频道创建存根

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

我写了一个装饰函数， ：

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)

由此，我想从未实现的原始文件中创建所有方法：

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

在下面的类的最终用例中，将存根放入方法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"

我遇到了此错误，我认为这意味着频道不再打开，但我不确定如何做得更好，以确保所有用户都使用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"

原始文件是：

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;
}

下面是我试图在元素添加装饰器后试图模拟的类。

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!"))

我通过传统上添加装饰仪进一步测试了这一点，这也有效：

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!"))

任何帮助将不胜感激。

Answer 1

您在代码的这一部分中遇到了问题，您没有设置预期的proto对象，而是您正在设置字符串，

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"

正确的方法是更改​​行stub.analysesentiment（text），使用

stub.AnalyseSentiment(SentimentRequest(text=text))