Skip to content

GRPC:Python

gRPC 의 python 버전.

Generate in python code

우선 패키지를 설치한다. 

pip install grpcio-tools

그리고 다음과 같이 코딩한다. 

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 

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 

# 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

See also