为什么在文件calculator_pb2.py中我找不到OperationRequest和OperationResponse类的实现，我应该在哪里寻找它们？

问题描述 投票：0回答：1

我为我的 grpc 服务创建了一个 .proto 文件：

syntax = "proto3";

service Calculator {
  rpc Add (OperationRequest) returns (OperationResponse);
  rpc Mul (OperationRequest) returns (OperationResponse);
  rpc Sub (OperationRequest) returns (OperationResponse);
  rpc Div (OperationRequest) returns (OperationResponse);
  rpc Pow (OperationRequest) returns (OperationResponse);
}

message OperationRequest {
  double a = 1;
  double b = 2; 
}

message OperationResponse {
  double result = 1;
}

然后生成文件calculator_pb2.py：

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: calculator.proto
# Protobuf Python Version: 5.28.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
    _runtime_version.Domain.PUBLIC,
    5,
    28,
    1,
    '',
    'calculator.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10\x63\x61lculator.proto\"(\n\x10OperationRequest\x12\t\n\x01\x61\x18\x01 \x01(\x01\x12\t\n\x01\x62\x18\x02 \x01(\x01\"#\n\x11OperationResponse\x12\x0e\n\x06result\x18\x01 \x01(\x01\x32\xf2\x01\n\nCalculator\x12,\n\x03\x41\x64\x64\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Mul\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Sub\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03\x44iv\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Pow\x12\x11.OperationRequest\x1a\x12.OperationResponseb\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'calculator_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  DESCRIPTOR._loaded_options = None
  _globals['_OPERATIONREQUEST']._serialized_start=20
  _globals['_OPERATIONREQUEST']._serialized_end=60
  _globals['_OPERATIONRESPONSE']._serialized_start=62
  _globals['_OPERATIONRESPONSE']._serialized_end=97
  _globals['_CALCULATOR']._serialized_start=100
  _globals['_CALCULATOR']._serialized_end=342
# @@protoc_insertion_point(module_scope)

和calculator_pb2_grpc.py：

# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings

import calculator_pb2 as calculator__pb2

GRPC_GENERATED_VERSION = '1.68.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False

try:
    from grpc._utilities import first_version_is_lower
    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
    _version_not_supported = True

if _version_not_supported:
    raise RuntimeError(
        f'The grpc package installed is at version {GRPC_VERSION},'
        + f' but the generated code in calculator_pb2_grpc.py depends on'
        + f' grpcio>={GRPC_GENERATED_VERSION}.'
        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
    )


class CalculatorStub(object):
    """Missing associated documentation comment in .proto file."""

    def __init__(self, channel):
        """Constructor.

        Args:
            channel: A grpc.Channel.
        """
        self.Add = channel.unary_unary(
                '/Calculator/Add',
                request_serializer=calculator__pb2.OperationRequest.SerializeToString,
                response_deserializer=calculator__pb2.OperationResponse.FromString,
                _registered_method=True)
        self.Mul = channel.unary_unary(
                '/Calculator/Mul',
                request_serializer=calculator__pb2.OperationRequest.SerializeToString,
                response_deserializer=calculator__pb2.OperationResponse.FromString,
                _registered_method=True)
        self.Sub = channel.unary_unary(
                '/Calculator/Sub',
                request_serializer=calculator__pb2.OperationRequest.SerializeToString,
                response_deserializer=calculator__pb2.OperationResponse.FromString,
                _registered_method=True)
        self.Div = channel.unary_unary(
                '/Calculator/Div',
                request_serializer=calculator__pb2.OperationRequest.SerializeToString,
                response_deserializer=calculator__pb2.OperationResponse.FromString,
                _registered_method=True)
        self.Pow = channel.unary_unary(
                '/Calculator/Pow',
                request_serializer=calculator__pb2.OperationRequest.SerializeToString,
                response_deserializer=calculator__pb2.OperationResponse.FromString,
                _registered_method=True)


class CalculatorServicer(object):
    """Missing associated documentation comment in .proto file."""

    def Add(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def Mul(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def Sub(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def Div(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def Pow(self, request, context):
        """Missing associated documentation comment in .proto file."""
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')


def add_CalculatorServicer_to_server(servicer, server):
    rpc_method_handlers = {
            'Add': grpc.unary_unary_rpc_method_handler(
                    servicer.Add,
                    request_deserializer=calculator__pb2.OperationRequest.FromString,
                    response_serializer=calculator__pb2.OperationResponse.SerializeToString,
            ),
            'Mul': grpc.unary_unary_rpc_method_handler(
                    servicer.Mul,
                    request_deserializer=calculator__pb2.OperationRequest.FromString,
                    response_serializer=calculator__pb2.OperationResponse.SerializeToString,
            ),
            'Sub': grpc.unary_unary_rpc_method_handler(
                    servicer.Sub,
                    request_deserializer=calculator__pb2.OperationRequest.FromString,
                    response_serializer=calculator__pb2.OperationResponse.SerializeToString,
            ),
            'Div': grpc.unary_unary_rpc_method_handler(
                    servicer.Div,
                    request_deserializer=calculator__pb2.OperationRequest.FromString,
                    response_serializer=calculator__pb2.OperationResponse.SerializeToString,
            ),
            'Pow': grpc.unary_unary_rpc_method_handler(
                    servicer.Pow,
                    request_deserializer=calculator__pb2.OperationRequest.FromString,
                    response_serializer=calculator__pb2.OperationResponse.SerializeToString,
            ),
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'Calculator', rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))
    server.add_registered_method_handlers('Calculator', rpc_method_handlers)


 # This class is part of an EXPERIMENTAL API.
class Calculator(object):
    """Missing associated documentation comment in .proto file."""

    @staticmethod
    def Add(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/Calculator/Add',
            calculator__pb2.OperationRequest.SerializeToString,
            calculator__pb2.OperationResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def Mul(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/Calculator/Mul',
            calculator__pb2.OperationRequest.SerializeToString,
            calculator__pb2.OperationResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def Sub(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/Calculator/Sub',
            calculator__pb2.OperationRequest.SerializeToString,
            calculator__pb2.OperationResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def Div(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/Calculator/Div',
            calculator__pb2.OperationRequest.SerializeToString,
            calculator__pb2.OperationResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def Pow(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(
            request,
            target,
            '/Calculator/Pow',
            calculator__pb2.OperationRequest.SerializeToString,
            calculator__pb2.OperationResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

这是生成命令：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. calculator.proto

然后我创建了 client.py ，我可以在其中使用

OperationRequest(a=5, b=3)
：

import grpc
import calculator_pb2
import calculator_pb2_grpc

def run():
    with grpc.insecure_channel("localhost:8080") as channel:
        stub = calculator_pb2_grpc.CalculatorStub(channel)

        print("Add: ",  )
        print("Mul: ", stub.Mul(calculator_pb2.OperationRequest(a=5, b=3)).result)
        print("Sub: ", stub.Sub(calculator_pb2.OperationRequest(a=5, b=3)).result)
        try:
            print("Div: ", stub.Div(calculator_pb2.OperationRequest(a=6, b=0)).result)
        except grpc.RpcError as e:
            print(f"Error during Div: {e.details()}")

        print("Pow: ", stub.Pow(calculator_pb2.OperationRequest(a=5, b=3)).result)

if __name__ == "__main__":
    run()

这是我的 server.py 使用 

OperationResponse(result=request.a + request.b)
:

import grpc
from concurrent import futures
import calculator_pb2
import calculator_pb2_grpc

class CalculatorServicer(calculator_pb2_grpc.CalculatorServicer):
    def Add(self, request, context):
        return calculator_pb2.OperationResponse(result=request.a + request.b)

    def Mul(self, request, context):
        return calculator_pb2.OperationResponse(result=request.a * request.b)

    def Sub(self, request, context):
        return calculator_pb2.OperationResponse(result=request.a - request.b)

    def Div(self, request, context):
        if request.b == 0:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Division by zero")
        return calculator_pb2.OperationResponse(result=request.a / request.b)

    def Pow(self, request, context):
        return calculator_pb2.OperationResponse(result=request.a ** request.b)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=3))

    calculator_pb2_grpc.add_CalculatorServicer_to_server(CalculatorServicer(), server)

    server.add_insecure_port("[::]:8080")
    print("Server is running on port 8080")

    server.start()

    server.wait_for_termination()

if __name__ == "__main__":
    serve()

calculator_pb2.py中没有这些类的定义，但服务工作正常

我查看了文件calculator_pb2.py和calculator_pb2_grpc.py，其中没有这些类的实现。 我也尝试询问ChatGPT，但没有得到明确的答案。但我想找到它们（确切的实现）

python protocol-buffers grpc stub
1个回答
0
投票

这些类是在运行时生成的

您的 .proto 生成了什么？

当您在 .proto 上运行协议缓冲区编译器时，编译器会以您选择的语言生成代码，您需要使用文件中描述的消息类型，包括获取和设置字段值、序列化消息到输出流，并从输入流解析消息。

Python 略有不同 - Python 编译器生成一个模块，其中包含 .proto 中每种消息类型的静态描述符，然后将其与元类一起使用以在运行时创建必要的 Python 数据访问类。

DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10\x63\x61lculator.proto\"(\n\x10OperationRequest\x12\t\n\x01\x61\x18\x01 \x01(\x01\x12\t\n\x01\x62\x18\x02 \x01(\x01\"#\n\x11OperationResponse\x12\x0e\n\x06result\x18\x01 \x01(\x01\x32\xf2\x01\n\nCalculator\x12,\n\x03\x41\x64\x64\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Mul\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Sub\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03\x44iv\x12\x11.OperationRequest\x1a\x12.OperationResponse\x12,\n\x03Pow\x12\x11.OperationRequest\x1a\x12.OperationResponseb\x06proto3')
最新问题
© www.soinside.com 2019 - 2025. All rights reserved.