Buf を使ってgRPCコトハジメ

この記事について

この記事は3-shake Advent Calendar 2024の20日目の記事です。
最近、Buf + Protobufを用いたgRPCのアプリケーションを開発しているので、そちらについて書いていきます。
個人的にgRPCを使った開発が初めてだったので、メモ/備忘録がてらでもあります。

そもそもgRPCってなんすか？

• Googleが開発した高速かつ効率的なRPC（Remote Procedure Call）フレームワーク
• サーバーとクライアントが事前に定義されたサービス仕様（Protocol Buffersを使用）に基づいて通信を行う
• 特徴
  • HTTP/2を基盤とした双方向ストリーミング通信のサポート
  • バイナリデータを使用する効率的なシリアライズ（Protocol Buffers）
  • 多言語対応

Bufってなに？

• Bufは、Protocol Buffers（Protobuf）のモダンなツールセット
• 特徴
  • .proto ファイルのコード生成とLintチェックを統一的に管理
  • 複数の言語に対応したコード生成がシュッとできる

サンプル用のプロジェクトの構成

.
├── protos
└── python_project

• Pythonをバックエンドとして使うようにしている
• protos ディレクトリにBuf関連の設定ファイルをまとめておく

Buf CLIのインストール

• mac OSの場合

% brew install bufbuild/buf/buf
% buf --version
1.47.2

• Linux, Windowsの場合はインストールガイドを元に提供されているバイナリをダウンロードしてPATHに設定

buf.yamlの作成

以下のコマンドでbufの初期化を行うと

% buf config init

このようなbuf.yamlが作成されます。

protos/buf.yaml

# For details on buf.yaml configuration, visit https://buf.build/docs/configuration/v2/buf-yaml
version: v2
lint:
  use:
    - STANDARD
breaking:
  use:
    - FILE

詳細はこちらにある通り、オプションの指定や、モジュールの導入などを行うことも可能です。
一旦、このままで進めてみましょう。

.protoの作成

以下のようにprotoファイルを作成する

protos/buf_sample_pj/ping/v1/ping.proto

syntax = "proto3";

package buf_sample_pj.ping.v1;

message PingRequest {
  string message = 1;
}

message PingResponse {
  string message = 1;
}

service PingService {
  rpc Get(PingRequest) returns (PingResponse);
}

この状態で buf lint を実行すると以下のようにBufで用意されているlint ruleをもとに警告が出されます。

buf_sample_pj/ping/v1/ping.proto:14:11:RPC request type "PingRequest" should be named "GetRequest" or "PingServiceGetRequest".
buf_sample_pj/ping/v1/ping.proto:14:33:RPC response type "PingResponse" should be named "GetResponse" or "PingServiceGetResponse".

警告を元に修正してみます。

protos/buf_sample_pj/ping/v1/ping.proto

syntax = "proto3";

package buf_sample_pj.ping.v1;

message PingServiceGetRequest {
  string message = 1;
}

message PingServiceGetResponse {
  string message = 1;
}

service PingService {
  rpc Get(PingServiceGetRequest) returns (PingServiceGetResponse);
}

buf.gen.yamlの作成

開発言語のインターフェイスファイルを生成するための設定ファイル、buf.gen.yamlを作成する。
今回はPython用に設定しています。

version: v1
plugins:
  - plugin: python
    out: ../python_project/protos
  - plugin: buf.build/protocolbuffers/pyi:v28.3
    out: ../python_project/protos
  - plugin: buf.build/grpc/python:v1.67.1
    out: ../python_project/protos

各フィールドの詳細は以下の通り

• version: v2
  • buf.gen.yaml ファイルのバージョンを指定
  • Buf のコード生成に使用する仕様のバージョン
• plugins
  • コード生成時に使用するプラグインをリストで指定
  • 各プラグインの設定項目
    • plugin: 使用するプラグインの名前またはリモートリポジトリでのパスを指定
    • out: 生成されたコードの出力先ディレクトリを指定
    • opt: プラグインに渡す追加オプションを指定(今回は未設定)
  • 各プラグインの詳細
    • plugin: python
      • Python 用のコードを生成する標準プラグイン
    • plugin: buf.build/protocolbuffers/pyi:v28.3
      • Python の型ヒント (.pyi ファイル) を生成するプラグイン
    • plugin: buf.build/grpc/python:v1.67.1
      • Python 用の gRPC サービスコードを生成するプラグイン

buf generateでコードを生成する

protobufをインストール後、

% brew install protobuf

以下のようなシェルを作成して、

build.sh

#!/bin/bash

rm -rf ../python_project/protos/*

buf generate

find ../python_project/protos/ -type d -exec touch {}/__init__.py \;

実行してみます。

% ./build.sh

生成されるコード

protos/buf_sample_pj/ping/v1/ping_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

from protos.ping.v1 import ping_pb2 as protos_dot_ping_dot_v1_dot_ping__pb2


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

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

        Args:
            channel: A grpc.Channel.
        """
        self.Get = channel.unary_unary(
                '/ping.v1.PingService/Get',
                request_serializer=protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetRequest.SerializeToString,
                response_deserializer=protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetResponse.FromString,
                _registered_method=True)


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

    def Get(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_PingServiceServicer_to_server(servicer, server):
    rpc_method_handlers = {
            'Get': grpc.unary_unary_rpc_method_handler(
                    servicer.Get,
                    request_deserializer=protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetRequest.FromString,
                    response_serializer=protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetResponse.SerializeToString,
            ),
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'ping.v1.PingService', rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))
    server.add_registered_method_handlers('ping.v1.PingService', rpc_method_handlers)


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

    @staticmethod
    def Get(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,
            '/ping.v1.PingService/Get',
            protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetRequest.SerializeToString,
            protos_dot_ping_dot_v1_dot_ping__pb2.PingServiceGetResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

protos/buf_sample_pj/ping/v1/ping_pb2.py

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: protos/ping/v1/ping.proto
# Protobuf Python Version: 5.29.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,
    29,
    1,
    '',
    'protos/ping/v1/ping.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x19protos/ping/v1/ping.proto\x12\x07ping.v1\"1\n\x15PingServiceGetRequest\x12\x18\n\x07message\x18\x01 \x01(\tR\x07message\"2\n\x16PingServiceGetResponse\x12\x18\n\x07message\x18\x01 \x01(\tR\x07message2U\n\x0bPingService\x12\x46\n\x03Get\x12\x1e.ping.v1.PingServiceGetRequest\x1a\x1f.ping.v1.PingServiceGetResponseb\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'protos.ping.v1.ping_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  DESCRIPTOR._loaded_options = None
  _globals['_PINGSERVICEGETREQUEST']._serialized_start=38
  _globals['_PINGSERVICEGETREQUEST']._serialized_end=87
  _globals['_PINGSERVICEGETRESPONSE']._serialized_start=89
  _globals['_PINGSERVICEGETRESPONSE']._serialized_end=139
  _globals['_PINGSERVICE']._serialized_start=141
  _globals['_PINGSERVICE']._serialized_end=226
# @@protoc_insertion_point(module_scope)

ping_pb2.pyi

from google.protobuf import descriptor as _descriptor
from google.protobuf import message as _message
from typing import ClassVar as _ClassVar, Optional as _Optional

DESCRIPTOR: _descriptor.FileDescriptor

class PingServiceGetRequest(_message.Message):
    __slots__ = ("message",)
    MESSAGE_FIELD_NUMBER: _ClassVar[int]
    message: str
    def __init__(self, message: _Optional[str] = ...) -> None: ...

class PingServiceGetResponse(_message.Message):
    __slots__ = ("message",)
    MESSAGE_FIELD_NUMBER: _ClassVar[int]
    message: str
    def __init__(self, message: _Optional[str] = ...) -> None: ...

生成されたコードを使ってgRPCサービスを実装する

必要なモジュールをインストール

今回はuvを使用しています。

% uv add grpcio
Resolved 2 packages in 357ms
Prepared 1 package in 890ms
Installed 1 package in 3ms
 + grpcio==1.68.1

% uv add protobuf
Resolved 3 packages in 269ms
Prepared 1 package in 149ms
Installed 1 package in 1ms
 + protobuf==5.29.2

% uv add grpcio-reflection
Resolved 4 packages in 196ms
Prepared 1 package in 69ms
Installed 1 package in 2ms
 + grpcio-reflection==1.68.1

Serviceの実装

BaseServiceという抽象クラスをまず実装してみましょう。
add_to_server という抽象メソッドでgrpcサーバへの登録をカプセル化しています。

python_project/services/base_service.py

from abc import ABC, abstractmethod

import grpc


class BaseService(ABC):
    @abstractmethod
    def add_to_server(self, server: grpc.aio.Server) -> None:
        pass

生成されたPingServiceServicerとBaseServiceと継承したPingServiceを実装。
具象メソッド add_to_server でServiceをgrpc.serverに追加できるようにしています。
また、protoで定義していた Get メソッドを実装。ここでは単純にリクエストで受け取ったmessageをそのまま返すだけに留めておきます。

python_project/services/ping.py

import grpc

from protos.ping.v1.ping_pb2 import PingServiceGetRequest, PingServiceGetResponse
from protos.ping.v1.ping_pb2_grpc import (
    PingServiceServicer,
    add_PingServiceServicer_to_server,
)
from services.base_service import BaseService


class PingService(PingServiceServicer, BaseService):
    def add_to_server(self, server: grpc.aio.server) -> None:
        add_PingServiceServicer_to_server(self, server)

    def Get(
        self, request: PingServiceGetRequest, context: grpc.aio.ServicerContext
    ) -> PingServiceGetResponse:
        return PingServiceGetResponse(message=request.message)

main.pyの実装

main.pyの実装は以下の通りです。

python

import logging
from concurrent import futures

import grpc
from grpc_reflection.v1alpha import reflection
from services.ping import PingService


def main():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    service_names = [
        reflection.SERVICE_NAME,
    ]

    services = [PingService()]
    for service in services:
        logging.info("Create service: %s", service.__class__.__name__)
        service.add_to_server(server)
        service_names.append(service.__class__.__name__)
    reflection.enable_server_reflection(service_names, server)

    bind_address = "[::]:50051"
    server.add_insecure_port(bind_address)
    logging.info("Starting server on %s", bind_address)
    server.start()
    logging.info("Server started, waiting for termination...")
    server.wait_for_termination()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()

ポイント

• service_names.append(service.__class__.__name__)
  • ここでサービス名を Reflection に追加している。これにより、grpcurlを使うことでサービス名が確認できる。
• reflection.enable_server_reflection(service_names, server)
  • Reflection機能を有効化。これにより、登録されたサービスを外部ツールが動的に検出可能になる。

まとめ

こんな感じでprotoファイルに定義したI/FをBufを用いてLintやコードの生成を行うことができました。
今回はPythonのサーバー側の実装だけに留めましたが、クラアント側のコードの生成も行うことができます。
Bufの導入で統一的なLintとコード生成できるので、良かったですね、という気持ち。