GRPC:Python
Generate in python code
우선 패키지를 설치한다.
그리고 다음과 같이 코딩한다.
import grpc_tools.protoc
import pkg_resources
def generate_protoc():
print("Generate protoc ...")
grpc_proto_include = pkg_resources.resource_filename("grpc_tools", "_proto")
recc_proto_include = os.path.join(RECC_DIR, "proto")
api_proto_path = os.path.join(recc_proto_include, "api.proto")
grpc_tools.protoc.main([
"grpc_tools.protoc",
f"-I{grpc_proto_include}",
f"-I{recc_proto_include}",
f"--python_out={recc_proto_include}",
f"--grpc_python_out={recc_proto_include}",
api_proto_path
])
Asyncio gRPC proto
- Github - grpc/examples/python/helloworld/ - 동기식, 기본.
- Github - grpc/examples/python/route_guide - 비동기식, Asyncio 사용.
syntax = "proto3";
package recc.node.proto;
service ReccApi {
rpc Test (TestQ) returns (TestA) {}
rpc ClientStreamingTest (stream TestQ) returns (TestA) {}
rpc ServerStreamingTest (TestQ) returns (stream TestA) {}
rpc BidirectionalStreamingTest (stream TestQ) returns (stream TestA) {}
}
message TestQ {
string msg = 1;
}
message TestA {
string msg = 1;
}
Asyncio gRPC server
# -*- coding: utf-8 -*-
import asyncio
import logging
import grpc
from typing import Optional, AsyncIterable
from recc.node.proto import api_pb2 as api
from recc.node.proto.api_pb2_grpc import (
ReccApiServicer,
add_ReccApiServicer_to_server,
)
DEFAULT_NODE_LISTEN_ADDRESS = "[::]:21000"
LOGGER_NAME_RECC_RPC = "recc.rpc"
logger = logging.getLogger(LOGGER_NAME_RECC_RPC)
class NodeRpcServicer(ReccApiServicer):
"""
RECC Node RPC Servicer.
"""
async def Test(
self,
request: api.TestQ,
context: grpc.aio.ServicerContext,
) -> api.TestA:
# logger.info(f"RPC/Test(msg={request.msg})")
print(f"RPC/Test(msg={request.msg})")
return api.TestA(msg=request.msg)
async def ClientStreamingTest(
self,
request_iterator: AsyncIterable[api.TestQ],
context: grpc.aio.ServicerContext,
) -> api.TestA:
total = ""
async for request in request_iterator:
print(f"RPC/ClientStreamingTest(msg={request.msg}) ...")
total += request.msg
print(f"RPC/ClientStreamingTest() Done -> {total}")
return api.TestA(msg=total)
async def ServerStreamingTest(
self,
request: api.TestQ,
context: grpc.aio.ServicerContext,
) -> AsyncIterable[api.TestA]:
print(f"RPC/ServerStreamingTest(msg={request.msg})")
for i in range(10):
generated_msg = f"{request.msg}[{i}]"
print(f"RPC/ServerStreamingTest() Streaming -> {generated_msg}")
yield api.TestA(msg=f"{generated_msg}")
print("RPC/ServerStreamingTest() Done")
async def BidirectionalStreamingTest(
self,
request_iterator: AsyncIterable[api.TestQ],
context: grpc.aio.ServicerContext,
) -> AsyncIterable[api.TestA]:
total = ""
async for request in request_iterator:
print(f"RPC/BidirectionalStreamingTest() Streaming -> {request.msg}")
total += request.msg
yield api.TestA(msg=f"{request.msg}")
print(f"RPC/BidirectionalStreamingTest() Last streaming -> {total}")
yield api.TestA(msg=total)
print("RPC/BidirectionalStreamingTest() Done")
async def run_node_server(
listen_address=DEFAULT_NODE_LISTEN_ADDRESS,
grace: Optional[float] = None,
) -> None:
server = grpc.aio.server()
add_ReccApiServicer_to_server(NodeRpcServicer(), server)
server.add_insecure_port(listen_address)
await server.start()
try:
await server.wait_for_termination()
except KeyboardInterrupt:
# Shuts down the server with 0 seconds of grace period. During the
# grace period, the server won't accept new connections and allow
# existing RPCs to continue within the grace period.
await server.stop(grace)
def main():
asyncio.run(run_node_server())
if __name__ == "__main__":
main()
Asyncio gRPC client
# -*- coding: utf-8 -*-
import asyncio
import grpc
from recc.node.proto import api_pb2 as api
from recc.node.proto.api_pb2_grpc import (
ReccApiStub,
)
DEFAULT_NODE_ADDRESS = "localhost:21000"
def generate_test_q():
for i in range(10):
yield api.TestQ(msg=f"Q[{i}]")
async def guide_list(stub: ReccApiStub) -> None:
async for response in stub.ServerStreamingTest(api.TestQ(msg="Q")):
print(f"SS -> TestA({response.msg})")
async def guide_route_chat(stub: ReccApiStub) -> None:
async for response in stub.BidirectionalStreamingTest(generate_test_q()):
print(f"BS -> {response.msg}")
async def test_node_client(address=DEFAULT_NODE_ADDRESS) -> None:
async with grpc.aio.insecure_channel(address) as channel:
stub = ReccApiStub(channel)
response = await stub.Test(api.TestQ(msg="Hello Node !"))
print(f"TestA({response.msg})")
response = await stub.ClientStreamingTest(generate_test_q())
print(f"CS -> TestA({response.msg})")
await guide_list(stub)
await guide_route_chat(stub)
def main():
asyncio.run(test_node_client())
if __name__ == "__main__":
main()
최대 메시지 크기 설정
클라이언트 측 설정:
import grpc
channel = grpc.insecure_channel(
'localhost:50051',
options=[
('grpc.max_send_message_length', 100 * 1024 * 1024), # 100MB 전송 가능
('grpc.max_receive_message_length', 100 * 1024 * 1024) # 100MB 수신 가능
]
)
stub = YourServiceStub(channel)
서버 측 설정:
from concurrent import futures
server = grpc.server(
futures.ThreadPoolExecutor(max_workers=10),
options=[
('grpc.max_send_message_length', 100 * 1024 * 1024),
('grpc.max_receive_message_length', 100 * 1024 * 1024)
]
)
server.add_insecure_port('[::]:50051')
server.start()
Testing
#!/usr/bin/env python
# coding=utf-8
import unittest
from grpc import StatusCode
from grpc_testing import server_from_dictionary, strict_real_time
from proto import rpc_pb2
class TestCase(unittest.TestCase):
def __init__(self, methodName) -> None:
super().__init__(methodName)
servicers = {
rpc_pb2.DESCRIPTOR.services_by_name['Library']: LibraryServicer()
}
self.test_server = server_from_dictionary(
servicers, strict_real_time())
def test_search(self):
request = rpc_pb2.Request(
id=2,
)
method = self.test_server.invoke_unary_unary(
method_descriptor=(rpc_pb2.DESCRIPTOR
.services_by_name['Library']
.methods_by_name['Search']),
invocation_metadata={},
request=request, timeout=1)
response, metadata, code, details = method.termination()
self.assertTrue(bool(response.status))
self.assertEqual(code, StatusCode.OK)
if __name__ == '__main__':
unittest.main()
Generated-code
service FortuneTeller {
// Returns the horoscope and zodiac sign for the given month and day.
rpc TellFortune(HoroscopeRequest) returns (HoroscopeResponse) {
// errors: invalid month or day, fortune unavailable
}
// Replaces the fortune for the given zodiac sign with the provided one.
rpc SuggestFortune(SuggestionRequest) returns (SuggestionResponse) {
// errors: invalid zodiac sign
}
}
When the service is compiled, the gRPC protoc
plugin generates code similar to the following _pb2_grpc.py
file:
import grpc
import fortune_pb2
class FortuneTellerStub(object):
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.TellFortune = channel.unary_unary(
'/example.FortuneTeller/TellFortune',
request_serializer=fortune_pb2.HoroscopeRequest.SerializeToString,
response_deserializer=fortune_pb2.HoroscopeResponse.FromString,
)
self.SuggestFortune = channel.unary_unary(
'/example.FortuneTeller/SuggestFortune',
request_serializer=fortune_pb2.SuggestionRequest.SerializeToString,
response_deserializer=fortune_pb2.SuggestionResponse.FromString,
)
class FortuneTellerServicer(object):
def TellFortune(self, request, context):
"""Returns the horoscope and zodiac sign for the given month and day.
errors: invalid month or day, fortune unavailable
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def SuggestFortune(self, request, context):
"""Replaces the fortune for the given zodiac sign with the provided
one.
errors: invalid zodiac sign
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_FortuneTellerServicer_to_server(servicer, server):
rpc_method_handlers = {
'TellFortune': grpc.unary_unary_rpc_method_handler(
servicer.TellFortune,
request_deserializer=fortune_pb2.HoroscopeRequest.FromString,
response_serializer=fortune_pb2.HoroscopeResponse.SerializeToString,
),
'SuggestFortune': grpc.unary_unary_rpc_method_handler(
servicer.SuggestFortune,
request_deserializer=fortune_pb2.SuggestionRequest.FromString,
response_serializer=fortune_pb2.SuggestionResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'example.FortuneTeller', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
Request timeout
- Stackoverflow - python grpc: setting timeout per grpc call
- Github - grpc/src/python/grpcio_tests/tests_aio/unit/timeout_test.py
# Copyright 2020 The gRPC Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests behavior of the timeout mechanism on client side."""
import asyncio
import logging
import platform
import random
import unittest
import datetime
import grpc
from grpc.experimental import aio
from tests_aio.unit._test_base import AioTestBase
from tests_aio.unit import _common
_SLEEP_TIME_UNIT_S = datetime.timedelta(seconds=1).total_seconds()
_TEST_SLEEPY_UNARY_UNARY = '/test/Test/SleepyUnaryUnary'
_TEST_SLEEPY_UNARY_STREAM = '/test/Test/SleepyUnaryStream'
_TEST_SLEEPY_STREAM_UNARY = '/test/Test/SleepyStreamUnary'
_TEST_SLEEPY_STREAM_STREAM = '/test/Test/SleepyStreamStream'
_REQUEST = b'\x00\x00\x00'
_RESPONSE = b'\x01\x01\x01'
async def _test_sleepy_unary_unary(unused_request, unused_context):
await asyncio.sleep(_SLEEP_TIME_UNIT_S)
return _RESPONSE
async def _test_sleepy_unary_stream(unused_request, unused_context):
yield _RESPONSE
await asyncio.sleep(_SLEEP_TIME_UNIT_S)
yield _RESPONSE
async def _test_sleepy_stream_unary(unused_request_iterator, context):
assert _REQUEST == await context.read()
await asyncio.sleep(_SLEEP_TIME_UNIT_S)
assert _REQUEST == await context.read()
return _RESPONSE
async def _test_sleepy_stream_stream(unused_request_iterator, context):
assert _REQUEST == await context.read()
await asyncio.sleep(_SLEEP_TIME_UNIT_S)
await context.write(_RESPONSE)
_ROUTING_TABLE = {
_TEST_SLEEPY_UNARY_UNARY:
grpc.unary_unary_rpc_method_handler(_test_sleepy_unary_unary),
_TEST_SLEEPY_UNARY_STREAM:
grpc.unary_stream_rpc_method_handler(_test_sleepy_unary_stream),
_TEST_SLEEPY_STREAM_UNARY:
grpc.stream_unary_rpc_method_handler(_test_sleepy_stream_unary),
_TEST_SLEEPY_STREAM_STREAM:
grpc.stream_stream_rpc_method_handler(_test_sleepy_stream_stream)
}
class _GenericHandler(grpc.GenericRpcHandler):
def service(self, handler_call_details):
return _ROUTING_TABLE.get(handler_call_details.method)
async def _start_test_server():
server = aio.server()
port = server.add_insecure_port('[::]:0')
server.add_generic_rpc_handlers((_GenericHandler(),))
await server.start()
return f'localhost:{port}', server
class TestTimeout(AioTestBase):
async def setUp(self):
address, self._server = await _start_test_server()
self._client = aio.insecure_channel(address)
self.assertEqual(grpc.ChannelConnectivity.IDLE,
self._client.get_state(True))
await _common.block_until_certain_state(self._client,
grpc.ChannelConnectivity.READY)
async def tearDown(self):
await self._client.close()
await self._server.stop(None)
async def test_unary_unary_success_with_timeout(self):
multicallable = self._client.unary_unary(_TEST_SLEEPY_UNARY_UNARY)
call = multicallable(_REQUEST, timeout=2 * _SLEEP_TIME_UNIT_S)
self.assertEqual(_RESPONSE, await call)
self.assertEqual(grpc.StatusCode.OK, await call.code())
async def test_unary_unary_deadline_exceeded(self):
multicallable = self._client.unary_unary(_TEST_SLEEPY_UNARY_UNARY)
call = multicallable(_REQUEST, timeout=0.5 * _SLEEP_TIME_UNIT_S)
with self.assertRaises(aio.AioRpcError) as exception_context:
await call
rpc_error = exception_context.exception
self.assertEqual(grpc.StatusCode.DEADLINE_EXCEEDED, rpc_error.code())
async def test_unary_stream_success_with_timeout(self):
multicallable = self._client.unary_stream(_TEST_SLEEPY_UNARY_STREAM)
call = multicallable(_REQUEST, timeout=2 * _SLEEP_TIME_UNIT_S)
self.assertEqual(_RESPONSE, await call.read())
self.assertEqual(_RESPONSE, await call.read())
self.assertEqual(grpc.StatusCode.OK, await call.code())
async def test_unary_stream_deadline_exceeded(self):
multicallable = self._client.unary_stream(_TEST_SLEEPY_UNARY_STREAM)
call = multicallable(_REQUEST, timeout=0.5 * _SLEEP_TIME_UNIT_S)
self.assertEqual(_RESPONSE, await call.read())
with self.assertRaises(aio.AioRpcError) as exception_context:
await call.read()
rpc_error = exception_context.exception
self.assertEqual(grpc.StatusCode.DEADLINE_EXCEEDED, rpc_error.code())
async def test_stream_unary_success_with_timeout(self):
multicallable = self._client.stream_unary(_TEST_SLEEPY_STREAM_UNARY)
call = multicallable(timeout=2 * _SLEEP_TIME_UNIT_S)
await call.write(_REQUEST)
await call.write(_REQUEST)
self.assertEqual(grpc.StatusCode.OK, await call.code())
async def test_stream_unary_deadline_exceeded(self):
multicallable = self._client.stream_unary(_TEST_SLEEPY_STREAM_UNARY)
call = multicallable(timeout=0.5 * _SLEEP_TIME_UNIT_S)
with self.assertRaises(aio.AioRpcError) as exception_context:
await call.write(_REQUEST)
await call.write(_REQUEST)
await call
rpc_error = exception_context.exception
self.assertEqual(grpc.StatusCode.DEADLINE_EXCEEDED, rpc_error.code())
async def test_stream_stream_success_with_timeout(self):
multicallable = self._client.stream_stream(_TEST_SLEEPY_STREAM_STREAM)
call = multicallable(timeout=2 * _SLEEP_TIME_UNIT_S)
await call.write(_REQUEST)
self.assertEqual(_RESPONSE, await call.read())
self.assertEqual(grpc.StatusCode.OK, await call.code())
async def test_stream_stream_deadline_exceeded(self):
multicallable = self._client.stream_stream(_TEST_SLEEPY_STREAM_STREAM)
call = multicallable(timeout=0.5 * _SLEEP_TIME_UNIT_S)
with self.assertRaises(aio.AioRpcError) as exception_context:
await call.write(_REQUEST)
await call.read()
rpc_error = exception_context.exception
self.assertEqual(grpc.StatusCode.DEADLINE_EXCEEDED, rpc_error.code())
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
unittest.main(verbosity=2)
With server authentication SSL/TLS
우선 인증서는 openssl을 사용해 만든다.
Client:
import grpc
import helloworld_pb2
with open('roots.pem', 'rb') as f:
creds = grpc.ssl_channel_credentials(f.read())
channel = grpc.secure_channel('myservice.example.com:443', creds)
stub = helloworld_pb2.GreeterStub(channel)
Server:
import grpc
import helloworld_pb2
from concurrent import futures
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
with open('key.pem', 'rb') as f:
private_key = f.read()
with open('chain.pem', 'rb') as f:
certificate_chain = f.read()
server_credentials = grpc.ssl_server_credentials( ( (private_key, certificate_chain), ) )
# Adding GreeterServicer to server omitted
server.add_secure_port('myservice.example.com:443', server_credentials)
server.start()
# Server sleep omitted