I created a .proto file for my grpc service:
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;
}
Then generated the files 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)
And 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)
Here is generation command: python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. calculator.proto
Then I created client.py where I can use 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()
Here is my server.py that uses 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()
There is no definition of these classes in calculator_pb2.py,but service is working correctly
I've looked through the files calculator_pb2.py and calculator_pb2_grpc.py there is no implementation of these classes in them. I also tried asking ChatGPT, but didn't get a clear answer. But I want to find them (exactly implementation)
Those classes are generated at runtime
What’s Generated from Your .proto?
When you run the protocol buffer compiler on a .proto, the compiler generates the code in your chosen language you’ll need to work with the message types you’ve described in the file, including getting and setting field values, serializing your messages to an output stream, and parsing your messages from an input stream.
Python is a little different — the Python compiler generates a module with a static descriptor of each message type in your .proto, which is then used with a metaclass to create the necessary Python data access class at runtime.
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')
See also this awesome answer on python meta-classes.