LoginSignup
4
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@inetcpl(𝕂′)

Buf + python-betterprotoプラグインを使ってgRPCサービスを実装する

Last updated at Posted at 2025-03-25

はじめに

数年前に Python で gRPC サーバーを少し触ったことがあったのですが、当時はまだ gRPC そのものに馴染みがなかったことと、公式ツール（protoc）の使い勝手にやや難しさを感じた記憶がありました。

そんなこんなで最近になって再び gRPC を触る機会があったのですが、遅ればせながらこのタイミングで Buf というツールを知りました。
Buf が提供している Buf CLI を使うと.proto ファイルの管理が容易になるほか、lint や format が標準で実行できるなど、proto ファイルの品質管理や開発体験がかなり改善されていることを実感しました。

さらに、python-betterproto というコミュニティプラグインを組み合わせれば、protoc の生成コードよりもシンプル且つ慣習的な Python コードが生成されることも分かりました。

本記事は自分の備忘録も兼ねつつ、Buf と python-betterproto を使ったら gRPC の開発体験はどのように変わるのか を検証した記録です。また、この 2 つを組み合わせる場合にハマったところもあったのでそのあたりも共有できたらと思います。
同じように python + gRPC で試してみようかなと考えている方の参考になればうれしいです。

前提

  • 本記事で紹介する環境は Mac OS を利用します
  • 本記事で紹介する Python やライブラリ、ツールのバージョンは以下を利用します
    • Python: 3.13
    • python-betterproto: v2.0.0-beta7
    • Buf CLI: 1.50.1
  • 本記事は Buf および python-betterproto の紹介にフォーカスしているので、基本的な gRPC の説明（gRPC の概念や proto ファイルの書き方など）は省略しています

Buf とは？

Buf は、Protocol Buffers（protobuf）のためのモダンな開発ツールです。公式の protoc コマンドは protobuf のコード生成やコンパイルに使われますが、設定や管理が少し面倒でした。
Buf はこうした問題を解決するために Buf CLI を提供しています。

Buf CLI の主な特徴は以下です。

  • .proto ファイルの記述ルールをチェックする Lint 機能
  • .proto ファイルのフォーマットを統一する Format 機能
  • コード生成のための設定ファイル（buf.gen.yaml）による再現可能なコード生成機能
  • .proto ファイルの破壊的変更を検知し、後方互換性を維持する Breaking Change 検出機能

なお、Buf を初めて触る場合、まずは公式の Quickstart を試してみることをおすすめします。

以下の記事もとても参考になります（感謝です！）

python-betterproto とは？

python-betterproto は、Python 向けの新しい gRPC コードジェネレーターです。
公式の Google が提供する protoc によるコード生成には、Python で使う上で少々不便なところがありました。それらの課題を解消して Python らしいシンプルなコードを自動生成するために作られたようです。

protoc でコード生成する場合の具体的な課題・不便なポイントは以下であり、python-betterproto ではこれらの問題が解決されています。

（python-betterproto の README から引用）

Motivation

This project exists because I am unhappy with the state of the official Google protoc plugin for Python.

  • No async support (requires additional grpclib plugin)
  • No typing support or code completion/intelligence (requires additional mypy plugin)
  • No __init__.py module files get generated
  • Output is not importable
    • Import paths break in Python 3 unless you mess with sys.path
  • Bugs when names clash (e.g. codecs package)
  • Generated code is not idiomatic
    • Completely unreadable runtime code-generation
    • Much code looks like C++ or Java ported 1:1 to Python
    • Capitalized function names like HasField() and SerializeToString()
    • Uses SerializeToString() rather than the built-in __bytes__()
    • Special wrapped types don't use Python's None
    • Timestamp/duration types don't use Python's built-in datetime module

python-betterproto を使った場合の具体的な嬉しいポイントは後述します。

本記事で作成するソースコード

以下に公開しているのでよければ参考にしてください。

開発環境の準備

まずは gRPC 環境を作るために必要な準備を行います。Buf と python-betterproto を使うためには、それぞれのツールを事前にインストールしておく必要があります。

ディレクトリ構成の準備

本記事では次のようなディレクトリ構成で進めます。

.
├── proto         # protoファイルを置く場所
└── pet_service   # 架空のペットサービスプロジェクト
    └── src       # Buf CLIによる生成されるコード格納先およびgRPCのサービス・サーバーを実装するディレクトリ

以下のコマンドでディレクトリを作ります。

cd your-project-root
mkdir -p proto pet_service/src

Buf CLI のインストール

Buf CLI は Homebrew を使って簡単にインストールできます。

brew install bufbuild/buf/buf

インストールが完了したらバージョンを確認しておきます。

buf --version
1.50.1

Python の環境セットアップ（python-betterproto）

Python 環境は仮想環境を用意しておくと便利です。以下は uv を使った例ですが、好きな方法で仮想環境を作成してください。
そのうえで、python-betterproto を以下のようにインストールします。

cd pet_service
uv init
uv add "betterproto[compiler]==2.0.0b7"

この betterproto[compiler] をインストールすると、Buf から Python のコードを生成するために必要なツールが全て揃います。

注意

2025年3月時点において、Buf のコミュニティプラグインとして提供されている python-betterproto はバージョン 1.2.5 となります。ただし、v1.2.5 は、optional キーワードや gRPC サーバーを実装するための機能に対応していません。
これらを利用するためには Beta 版をインストールする必要があります。現時点での最新 Beta 版（v2.0.0b7）を利用するには上記のようにインストールすれば OK です。

(参考)
https://github.com/danielgtaylor/python-betterproto/issues/608

buf.yaml の作成と設定

Buf を使ってプロジェクトを管理するにはbuf.yaml という設定ファイルを作ります。このファイル内でプロジェクトで使う Lint や Breaking Change のルールなどを設定できます。
以下のコマンドを実行すると proto ディレクトリ配下に buf.yaml が生成されます。

cd proto
buf config init

buf.yamlが生成されたら、ファイルの内容を以下のように変更してみます。

# For details on buf.yaml configuration, visit https://buf.build/docs/configuration/v2/buf-yaml
version: v2
lint:
  use:
    - STANDARD
  ignore:
    - google/type/datetime.proto
breaking:
  use:
    - FILE
設定のポイントと補足
  • version: v2
    • 設定ファイルのバージョンを指定します。
  • lint: use: - STANDARD
    • Buf が推奨する標準の Lint ルールセットを使うことを指定しています。標準的な命名規則やベストプラクティスを自動的にチェックしてくれます。
  • lint: ignore
    • 指定した proto ファイル（この後の章で作成する google/type/datetime.proto）を lint チェックの対象外にします。
    • 今回は Google 公式のプロトファイルをそのまま使う（修正不要）ため、このファイルに対して lint エラーが出ないように除外しています。
  • breaking: use: - FILE
    • API の後方互換性を保つための Breaking Change チェックルールです。ファイルレベルで変更の影響を確認するために設定しています。
      • Breaking Change により、proto ファイルの変更による意図しない API の破壊的変更を防ぐことができます。

v2 の書き方の詳細は以下の記事を参照してください。

.proto ファイルの作成

gRPC サービスを構築するためには、まず.proto ファイルを作成して、サービスのインターフェースやメッセージの構造を定義する必要があります。
本記事では、独自サービスの.proto 定義（pet.proto）に加え、Google 公式の.proto 定義（datetime.proto）も利用してみます。

ディレクトリ構成の準備

事前準備で作成した proto ディレクトリ内に必要なファイルを配置します。最終的なディレクトリ構成は以下のようになります。

.
└── proto
    ├── buf.gen.yaml
    ├── buf.yaml
    ├── google
    │   └── type
    │       └── datetime.proto # <-これと
    └── pet
        └── v1
            └── pet.proto   # <-これを作る

datetime.proto の作成

google/type/datetime.protoは、googleapis公式の日時を表すためのメッセージ定義です。（後述の pet.proto 内で import して利用します）
ここでは、こちらの datetime.proto の内容をコピーし、ローカルの datetime.proto にペーストすれば OK です。

pet.protoの作成

次に、以下の内容で pet/v1/pet.proto を作成します。このファイルではペットストアのサービスとメッセージを定義します。

pet.proto
syntax = "proto3";

package pet.v1;

import "google/type/datetime.proto";

// PetType represents the different types of pets in the pet store.
enum PetType {
  PET_TYPE_UNSPECIFIED = 0;
  PET_TYPE_CAT = 1;
  PET_TYPE_dog = 2;
  PET_TYPE_SNAKE = 3;
  PET_TYPE_HAMSTER = 4;
}

// Pet represents a pet in the pet store.
message Pet {
        PetType pet_type = 1;
  string pet_id = 2;
  string name = 3;
  google.type.DateTime created_at = 4;
}

message GetPetRequest {
  string pet_id = 1;
}

message GetPetResponse {
  Pet pet = 1;
}

message PutPetRequest {
  PetType pet_type = 1;
  string name = 2;
}

message PutPetResponse {
  Pet pet = 1;
}

message DeletePetRequest {
  string pet_id = 1;
}

message DeletePetResponse {}

service PetStore {
  rpc GetPet(GetPetRequest) returns (GetPetResponse) {}
  rpc PutPet(PutPetRequest) returns (PutPetResponse) {}
  rpc DeletePet(DeletePetRequest) returns (DeletePetResponse) {}
}

以上で必要な.proto ファイルの作成は完了です。

buf lint の実行

.proto ファイルを作成できたので、Buf の便利機能の一つである buf lint を使ってさっき作成した.proto ファイルのチェックを行ってみます。

cd proto
buf lint

すると、次のようなエラーが表示されるはずです。

pet/v1/pet.proto:11:3:Enum value name "PET_TYPE_dog" should be UPPER_SNAKE_CASE, such as "PET_TYPE_DOG".
pet/v1/pet.proto:47:9:Service name "PetStore" should be suffixed with "Service".

以下のようにエラーを修正します。

pet.proto
-PET_TYPE_dog = 2;
+PET_TYPE_DOG = 2;

......

-service PetStore {
+service PetStoreService {

buf lint を実行すると、上記のように proto 定義の構文的なミス（前者）や命名規則から外れた部分（後者）を Style Guide に従って解析してくれます。
buf.yaml 内に Lint ルールのカスタマイズも定義できます。

buf format の実行

buf format も使ってみます。以下のように -w オプションを指定すればファイルを直接フォーマットしてくれます。

buf format -w

PetType pet_type = 1; のインデントがずれていましたが、format によって修正されるはずです。

proto ファイルから開発用コードを生成する

proto ファイルが用意できたら、Buf CLI を使って開発に必要な Python のコードを生成します。

buf.gen.yaml の作成と設定

まず、コード生成用の設定ファイル buf.gen.yaml を proto ディレクトリ直下に作成します。
後述する buf generate コマンドでは、この設定ファイルに基づいてコードを生成してくれます。
内容は以下のようになります。

version: v2
clean: true # コード生成時に、出力ディレクトリを事前にクリアする
plugins:
  - local: protoc-gen-python_betterproto # 利用するプラグイン名を指定（後述）
    out: ../pet_service/src/gen # コード生成先のディレクトリを指定
    strategy: all # プラグインを一括実行する設定（後述）
inputs:
  - directory: . # protoファイルの読み込み対象パスを指定

python-betterproto を使う上で重要なポイント

plugins.local: protoc-gen-python_betterproto について

plugins という項目は、proto ファイルから特定の言語（今回の場合は Python）向けのコードを生成する際に使用するプラグインを定義しています。
つまり、ここで指定したプラグインが、Buf CLI によって実行され、各言語に対応したコードが出力される仕組みとなります。

ここで注意が必要なのですが、本記事で使う python-betterproto は Buf CLI が公式にサポートするプラグインではなく、コミュニティプラグインです。
コミュニティプラグインは、Buf CLI において - remote: を指定することで Buf build 上から簡単に利用することができます。

例えば、python-betterproto については Buf build に次のような指定例が掲載されています。

- remote: buf.build/community/danielgtaylor-betterproto:v1.2.5
  out: gen

しかし現状、Buf build のコミュニティプラグインとして公開されているのは v1 系列のみで、本記事で使いたい Beta 版（v2 系統）はまだ公開されていません。
なので、- remote: 指定では v2 を利用することができません。

そこで現時点では、python-betterproto の最新 Beta 版を利用するために以下のような方法を取る必要があります。

  1. - local: 指定で buf にプラグインを認識させる
  2. 仮想環境内で必要な依存関係（python-betterproto v2）をインストールし、activate しておく
  3. buf generate を実行してコード生成をする

（参考）

strategy: all について

Buf CLI はデフォルトで、proto ファイルのあるディレクトリごとに個別にプラグインを実行します。そのため、本記事のように複数のディレクトリ（例えば、google と pet）にまたがって proto ファイルを用意すると、ディレクリごとに別々にプラグインが起動します。
その結果、datetime.proto が複数回コード生成されるなど、望ましくない重複が起きてしまいます。これを避けるために、strategy: all を指定しています。
strategy: all を指定することで、Buf は指定されたすべての入力ディレクトリを 1 つのまとまったモジュールとして扱い、プラグインの実行を 1 回にまとめてくれます。この設定があることでコードの重複生成を防ぐことができます。

buf generate の実行

コード生成用の設定ファイルを作成したので、実際に buf generate コマンドを実行してコードを生成してみます。
前述したとおり、以下の手順でコマンドを実行します。

２. 仮想環境内で必要な依存関係（python-betterproto v2）をインストールし、activate しておく
３. buf generate を実行してコード生成をする

cd proto
source ../pet_service/.venv/bin/activate
buf generate

正常に生成が完了すると、pet_service/src/gen ディレクトリ直下に以下のような Python コードが生成されます。

pet_service/src
└── gen
    ├── __init__.py
    ├── google
    │   ├── __init__.py
    │   └── type
    │       └── __init__.py
    └── pet
        ├── __init__.py
        └── v1
            └── __init__.py

python-betterproto で生成されたコードを見てみる

__init__.pyがたくさん生成されますが、proto から生成されたコード自体はそれぞれ以下に記述されています。

  • pet_service/src/gen/pet/v1/__init__.py
  • pet_service/src/gen/google/type/__init__.py

以下に python-betterproto を使用して生成されたコードの特徴と、公式の protoc との違い、個人的に気に入っているポイントを含めて紹介します。

__init__.py の自動生成

公式の protoc では、生成されたコードに __init__.py が含まれないため、Python パッケージとして認識させるには手動で __init__.py を追加する必要があります。
一方、python-betterproto は自動的に __init__.py を生成してくれるので追加の手間が省けます。

インポートパスの問題の解消

protoc では、生成されたコードのインポートパスが相対パスになります。生成先ディレクトリの構造次第にはなりますが、何も対策しないとインポートエラーが発生する辛みポイントがあります。

これに対し、python-betterproto は生成されたコードがどのディレクトリに配置されても適切にインポートできるように設計されています。

命名規則の改善

protoc で生成されるメソッド名は、例えば GetPet のようなパスカルケースであり、Python の一般的な命名規則とは異なります。
python-betterproto ではメソッド名がスネークケース（get_pet）で生成されるので、より Python らしいコードになります。

型ヒントのサポート

python-betterproto は、生成されるコードに型ヒントが含まれているので静的型チェックツール（mypy など） との相性が良くなっています。
なお、protoc では、pyi というプラグインを使えば型ヒント用のコード (.pyi ファイル) を生成することができますが、python-betterproto では生成されたコード自体に型ヒントが含まれています。

生成されるコードの可読性向上

公式の protoc で生成されたコードは、以下のように C++や Java をそのまま移植したような少しわかりづらいコードが生成されます。

pet_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 pet.v1 import pet_pb2 as pet_dot_v1_dot_pet__pb2


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

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

        Args:
            channel: A grpc.Channel.
        """
        self.GetPet = channel.unary_unary(
                '/pet.v1.PetStore/GetPet',
                request_serializer=pet_dot_v1_dot_pet__pb2.GetPetRequest.SerializeToString,
                response_deserializer=pet_dot_v1_dot_pet__pb2.GetPetResponse.FromString,
                _registered_method=True)
        self.PutPet = channel.unary_unary(
                '/pet.v1.PetStore/PutPet',
                request_serializer=pet_dot_v1_dot_pet__pb2.PutPetRequest.SerializeToString,
                response_deserializer=pet_dot_v1_dot_pet__pb2.PutPetResponse.FromString,
                _registered_method=True)
        self.DeletePet = channel.unary_unary(
                '/pet.v1.PetStore/DeletePet',
                request_serializer=pet_dot_v1_dot_pet__pb2.DeletePetRequest.SerializeToString,
                response_deserializer=pet_dot_v1_dot_pet__pb2.DeletePetResponse.FromString,
                _registered_method=True)


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

    def GetPet(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 PutPet(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 DeletePet(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_PetStoreServicer_to_server(servicer, server):
    rpc_method_handlers = {
            'GetPet': grpc.unary_unary_rpc_method_handler(
                    servicer.GetPet,
                    request_deserializer=pet_dot_v1_dot_pet__pb2.GetPetRequest.FromString,
                    response_serializer=pet_dot_v1_dot_pet__pb2.GetPetResponse.SerializeToString,
            ),
            'PutPet': grpc.unary_unary_rpc_method_handler(
                    servicer.PutPet,
                    request_deserializer=pet_dot_v1_dot_pet__pb2.PutPetRequest.FromString,
                    response_serializer=pet_dot_v1_dot_pet__pb2.PutPetResponse.SerializeToString,
            ),
            'DeletePet': grpc.unary_unary_rpc_method_handler(
                    servicer.DeletePet,
                    request_deserializer=pet_dot_v1_dot_pet__pb2.DeletePetRequest.FromString,
                    response_serializer=pet_dot_v1_dot_pet__pb2.DeletePetResponse.SerializeToString,
            ),
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'pet.v1.PetStore', rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))
    server.add_registered_method_handlers('pet.v1.PetStore', rpc_method_handlers)


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

    @staticmethod
    def GetPet(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,
            '/pet.v1.PetStore/GetPet',
            pet_dot_v1_dot_pet__pb2.GetPetRequest.SerializeToString,
            pet_dot_v1_dot_pet__pb2.GetPetResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def PutPet(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,
            '/pet.v1.PetStore/PutPet',
            pet_dot_v1_dot_pet__pb2.PutPetRequest.SerializeToString,
            pet_dot_v1_dot_pet__pb2.PutPetResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)

    @staticmethod
    def DeletePet(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,
            '/pet.v1.PetStore/DeletePet',
            pet_dot_v1_dot_pet__pb2.DeletePetRequest.SerializeToString,
            pet_dot_v1_dot_pet__pb2.DeletePetResponse.FromString,
            options,
            channel_credentials,
            insecure,
            call_credentials,
            compression,
            wait_for_ready,
            timeout,
            metadata,
            _registered_method=True)
pet_pb2.py 
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: pet/v1/pet.proto
# Protobuf Python Version: 6.30.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,
    6,
    30,
    1,
    '',
    'pet/v1/pet.proto'
)
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()


from google.type import datetime_pb2 as google_dot_type_dot_datetime__pb2


DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10pet/v1/pet.proto\x12\x06pet.v1\x1a\x1agoogle/type/datetime.proto\"\x92\x01\n\x03Pet\x12*\n\x08pet_type\x18\x01 \x01(\x0e\x32\x0f.pet.v1.PetTypeR\x07petType\x12\x15\n\x06pet_id\x18\x02 \x01(\tR\x05petId\x12\x12\n\x04name\x18\x03 \x01(\tR\x04name\x12\x34\n\ncreated_at\x18\x04 \x01(\x0b\x32\x15.google.type.DateTimeR\tcreatedAt\"&\n\rGetPetRequest\x12\x15\n\x06pet_id\x18\x01 \x01(\tR\x05petId\"/\n\x0eGetPetResponse\x12\x1d\n\x03pet\x18\x01 \x01(\x0b\x32\x0b.pet.v1.PetR\x03pet\"O\n\rPutPetRequest\x12*\n\x08pet_type\x18\x01 \x01(\x0e\x32\x0f.pet.v1.PetTypeR\x07petType\x12\x12\n\x04name\x18\x02 \x01(\tR\x04name\"/\n\x0ePutPetResponse\x12\x1d\n\x03pet\x18\x01 \x01(\x0b\x32\x0b.pet.v1.PetR\x03pet\")\n\x10\x44\x65letePetRequest\x12\x15\n\x06pet_id\x18\x01 \x01(\tR\x05petId\"\x13\n\x11\x44\x65letePetResponse*q\n\x07PetType\x12\x18\n\x14PET_TYPE_UNSPECIFIED\x10\x00\x12\x10\n\x0cPET_TYPE_CAT\x10\x01\x12\x10\n\x0cPET_TYPE_dog\x10\x02\x12\x12\n\x0ePET_TYPE_SNAKE\x10\x03\x12\x14\n\x10PET_TYPE_HAMSTER\x10\x04\x32\xc4\x01\n\x08PetStore\x12\x39\n\x06GetPet\x12\x15.pet.v1.GetPetRequest\x1a\x16.pet.v1.GetPetResponse\"\x00\x12\x39\n\x06PutPet\x12\x15.pet.v1.PutPetRequest\x1a\x16.pet.v1.PutPetResponse\"\x00\x12\x42\n\tDeletePet\x12\x18.pet.v1.DeletePetRequest\x1a\x19.pet.v1.DeletePetResponse\"\x00\x62\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'pet.v1.pet_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
  DESCRIPTOR._loaded_options = None
  _globals['_PETTYPE']._serialized_start=488
  _globals['_PETTYPE']._serialized_end=601
  _globals['_PET']._serialized_start=57
  _globals['_PET']._serialized_end=203
  _globals['_GETPETREQUEST']._serialized_start=205
  _globals['_GETPETREQUEST']._serialized_end=243
  _globals['_GETPETRESPONSE']._serialized_start=245
  _globals['_GETPETRESPONSE']._serialized_end=292
  _globals['_PUTPETREQUEST']._serialized_start=294
  _globals['_PUTPETREQUEST']._serialized_end=373
  _globals['_PUTPETRESPONSE']._serialized_start=375
  _globals['_PUTPETRESPONSE']._serialized_end=422
  _globals['_DELETEPETREQUEST']._serialized_start=424
  _globals['_DELETEPETREQUEST']._serialized_end=465
  _globals['_DELETEPETRESPONSE']._serialized_start=467
  _globals['_DELETEPETRESPONSE']._serialized_end=486
  _globals['_PETSTORE']._serialized_start=604
  _globals['_PETSTORE']._serialized_end=800
# @@protoc_insertion_point(module_scope)

一方、python-betterproto で生成されるコードは以下の内容となっており、Python 開発者が直感的に理解しやすいコードになっています。

pet_service/src/gen/pet/v1/__init__.py
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: pet/v1/pet.proto
# plugin: python-betterproto
# This file has been @generated

from dataclasses import dataclass
from typing import (
    TYPE_CHECKING,
    Dict,
    Optional,
)

import betterproto
import grpclib
from betterproto.grpc.grpclib_server import ServiceBase

from ...google import type as __google_type__


if TYPE_CHECKING:
    import grpclib.server
    from betterproto.grpc.grpclib_client import MetadataLike
    from grpclib.metadata import Deadline


class PetType(betterproto.Enum):
    """PetType represents the different types of pets in the pet store."""

    UNSPECIFIED = 0
    CAT = 1
    DOG = 2
    SNAKE = 3
    HAMSTER = 4


@dataclass(eq=False, repr=False)
class Pet(betterproto.Message):
    """Pet represents a pet in the pet store."""

    pet_type: "PetType" = betterproto.enum_field(1)
    pet_id: str = betterproto.string_field(2)
    name: str = betterproto.string_field(3)
    created_at: "__google_type__.DateTime" = betterproto.message_field(4)


@dataclass(eq=False, repr=False)
class GetPetRequest(betterproto.Message):
    pet_id: str = betterproto.string_field(1)


@dataclass(eq=False, repr=False)
class GetPetResponse(betterproto.Message):
    pet: "Pet" = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class PutPetRequest(betterproto.Message):
    pet_type: "PetType" = betterproto.enum_field(1)
    name: str = betterproto.string_field(2)


@dataclass(eq=False, repr=False)
class PutPetResponse(betterproto.Message):
    pet: "Pet" = betterproto.message_field(1)


@dataclass(eq=False, repr=False)
class DeletePetRequest(betterproto.Message):
    pet_id: str = betterproto.string_field(1)


@dataclass(eq=False, repr=False)
class DeletePetResponse(betterproto.Message):
    pass


class PetStoreServiceStub(betterproto.ServiceStub):
    async def get_pet(
        self,
        get_pet_request: "GetPetRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetPetResponse":
        return await self._unary_unary(
            "/pet.v1.PetStoreService/GetPet",
            get_pet_request,
            GetPetResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def put_pet(
        self,
        put_pet_request: "PutPetRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PutPetResponse":
        return await self._unary_unary(
            "/pet.v1.PetStoreService/PutPet",
            put_pet_request,
            PutPetResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def delete_pet(
        self,
        delete_pet_request: "DeletePetRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "DeletePetResponse":
        return await self._unary_unary(
            "/pet.v1.PetStoreService/DeletePet",
            delete_pet_request,
            DeletePetResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class PetStoreServiceBase(ServiceBase):

    async def get_pet(self, get_pet_request: "GetPetRequest") -> "GetPetResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def put_pet(self, put_pet_request: "PutPetRequest") -> "PutPetResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def delete_pet(
        self, delete_pet_request: "DeletePetRequest"
    ) -> "DeletePetResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_get_pet(
        self, stream: "grpclib.server.Stream[GetPetRequest, GetPetResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_pet(request)
        await stream.send_message(response)

    async def __rpc_put_pet(
        self, stream: "grpclib.server.Stream[PutPetRequest, PutPetResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.put_pet(request)
        await stream.send_message(response)

    async def __rpc_delete_pet(
        self, stream: "grpclib.server.Stream[DeletePetRequest, DeletePetResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.delete_pet(request)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/pet.v1.PetStoreService/GetPet": grpclib.const.Handler(
                self.__rpc_get_pet,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetPetRequest,
                GetPetResponse,
            ),
            "/pet.v1.PetStoreService/PutPet": grpclib.const.Handler(
                self.__rpc_put_pet,
                grpclib.const.Cardinality.UNARY_UNARY,
                PutPetRequest,
                PutPetResponse,
            ),
            "/pet.v1.PetStoreService/DeletePet": grpclib.const.Handler(
                self.__rpc_delete_pet,
                grpclib.const.Cardinality.UNARY_UNARY,
                DeletePetRequest,
                DeletePetResponse,
            ),
        }

また、上記のように protoc の場合は xxxx_pb2_grpc.py と xxxx_pb2.py という役割の異なる 2 つのファイルを生成する必要がありますが、python-betterproto は 1 ファイルに収まってる点も気に入っています。

非同期サポート

async/await を利用した非同期処理が可能で、非同期の gRPC クライアントやサーバーの実装が簡単にできます。

gRPC サービスの実装

proto から Python コードを生成できたので、実際にサービスを実装します。

pet_service
├── src
│   ├── gen
│   └── services
│       ├── __init__.py
│       └── pet.py   # ここにpetサービスを実装
└── proto

サービスの実装は、python-betterproto で生成されたサービスの基底クラスを継承し、各 RPC メソッドをオーバーライドするだけです。

以下は、get_pet メソッドを実際する場合の例です。今回は非同期に対応した gRPC サービスにするので async メソッドにしています。

# python-betterprotoで生成されたクラスをインポートする
from gen.pet.v1 import (
    PetStoreServiceBase,
    GetPetRequest,
    GetPetResponse,
    Pet,
    PetType,
)
from gen.google.type import DateTime


class PetServiceImpl(PetStoreServiceBase):
    async def get_pet(self, get_pet_request: GetPetRequest) -> GetPetResponse:
        print(f"get_pet: {get_pet_request.pet_id}")

        # 実際はここでDBなどからデータを取得するが、今回はダミーデータを返す
        dummy_pet = Pet(
            pet_id="1",
            pet_type=PetType(PetType.DOG),
            name="Buddy",
            created_at=DateTime(year=2021, month=8, day=1, hours=12),
        )
        return GetPetResponse(pet=dummy_pet)

gRPC サーバーの実装

次に、作成したサービスを公開するための gRPC サーバーを実装します。

pet_service
├── src
│   ├── gen
│   ├── services
│   │   ├── __init__.py
│   │   └── pet.py
│   └── server.py  # gRPCサーバー実装
└── proto

サーバーの実装は、grpclib という非同期対応の gRPC ライブラリを利用します。
※ grpclib は python-betterproto の依存パッケージとしてインストールされています。

以下は、grpclib を使って先ほど作成したサービスを起動しています。

from grpclib.server import Server
from services.pet import PetServiceImpl
import asyncio


async def start_server(host: str = "127.0.0.1", port: int = 50051) -> None:
    server = Server([PetServiceImpl()]) # サービスの登録は Server にサービス実装を渡すだけ
    await server.start(host, port)
    await server.wait_closed()

if __name__ == "__main__":
    asyncio.run(start_server())

gRPC クライアントの実装

サーバーの実装ができたら、次は gRPC サービスに接続するクライアントを実装します。

pet_service
├── src
│   ├── gen
│   ├── services
│   │   ├── __init__.py
│   │   └── pet.py
│   ├── server.py
│   └── client.py  # gRPCクライアント実装
└── proto

クライアント側でもシンプルな async/await を使ったコードで gRPC 通信できます。
以下はクライアントの実装例です。

import asyncio
from grpclib.client import Channel

from gen.pet.v1 import (
    PetStoreServiceStub,
    GetPetRequest,
)


async def client(host: str = "127.0.0.1", port: int = 50051) -> None:
    # grpclib.client.Channel でサーバーとの接続チャネルを作成
    channel = Channel(host, port)

    # 自動生成された PetStoreServiceStub を利用してget_petメソッドを呼び出す
    service = PetStoreServiceStub(channel)
    response = await service.get_pet(GetPetRequest(pet_id="123"))

    print(response)
    channel.close()

if __name__ == "__main__":
    asyncio.run(client())

通信してみる

サーバー用とクライアント用の 2 つのターミナルを立ち上げ、まずは以下のコマンドでサーバーを起動します。

サーバー用ターミナル
# pet_serviceディレクトリに移動しておきます
source .venv/bin/activate
python src/server.py

続いて、クライアントを立ち上げます。

クライアント用ターミナル
# pet_serviceディレクトリに移動しておきます
source .venv/bin/activate
python src/client.py

それぞれのターミナルに以下の結果が出力されれば通信成功です。

サーバー用ターミナル
get_pet: 123
クライアント用ターミナル
GetPetResponse(pet=Pet(pet_type=PetType.DOG, pet_id='1', name='Buddy', created_at=DateTime(year=2021, month=8, day=1, hours=12)))

なお、Buf CLI には通信テスト用の便利なコマンド buf curl が用意されています。これを使ってサーバーに通信することもできます。

# protoディレクトリに移動しておきます
buf curl --http2-prior-knowledge --schema pet/v1/pet.proto --data '{\"pet_id\": \"1\"}' --protocol grpc http://localhost:50051/pet.v1.PetStoreService/GetPet

# 結果
~/develop/grpc/buf-betterproto-sandbox/proto ~/develop/grpc/buf-betterproto-sandbox/pet_service
{
  "pet": {
    "petType": "PET_TYPE_DOG",
    "petId": "1",
    "name": "Buddy",
    "createdAt": {
      "year": 2021,
      "month": 8,
      "day": 1,
      "hours": 12
    }
  }
}
~/develop/grpc/buf-betterproto-sandbox/pet_service

buf curl の使い方の詳細は以下の公式ドキュメントを参考にしてください。

CI（ Github Actions ）で Buf CLI を実行する

Buf CLI は GitHub アクションも公開しています。

これを使った簡易的なワークフローの実装例は以下です。

.github/workflows/buf-ci.yaml
name: Buf CI
on:
  push:
    branches:
      - main
  pull_request:
  delete:
permissions:
  contents: read
  pull-requests: write
jobs:
  buf:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: bufbuild/buf-action@v1
        with:
          # protoファイルが置いてあるディレクトリを指定
          input: "proto"

          # PR上にBuf CLIの実行結果コメントを自動的に書き込むために必要
          github_token: "${{ secrets.GITHUB_TOKEN }}"

          # 破壊的変更（Breaking Changes）の検知を利用する設定
          # PRで変更したprotoファイルが、mainブランチ（マージ先）のprotoファイルに対して後方互換性を壊していないかをチェック
          breaking_against: "${{ github.event.repository.clone_url }}#format=git,commit=${{ github.event.pull_request.base.sha }},subdir=proto"

詳細な使い方は README および以下の公式ドキュメントを参考にしてください。

おわりに

数年ぶりに gRPC を触ってみて、Buf と python-betterproto による開発体験の向上を感じました。
特に、proto ファイルの管理が非常に楽になったことと、生成される Python コードがシンプルで読みやすくなった点がとても印象的でした。

懸念点としては、python-betterproto の開発が最近あまり活発ではなく、Buf の公式コミュニティプラグインの v2 対応も現時点では未対応な点が気になります。
また、python-betterproto は protoc と比較すると動作が遅いという報告もあるようです。

ただ、便利なツールであることは間違いないので、今後の動向にも期待してウォッチしていきたいと思います。

4
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?