Getting Started

Installation

Installation from PyPI is as simple as running:

python3 -m pip install -U betterproto

If you are using Windows, then the following should be used instead:

py -3 -m pip install -U betterproto

To include the protoc plugin, install betterproto[compiler] instead of betterproto, e.g.

python3 -m pip install -U "betterproto[compiler]"

Compiling proto files

Given you installed the compiler and have a proto file, e.g example.proto:

syntax = "proto3";

package hello;

// Greeting represents a message you can tell a user.
message Greeting {
  string message = 1;
}

To compile the proto you would run the following:

You can run the following to invoke protoc directly:

mkdir hello
protoc -I . --python_betterproto_out=lib example.proto

or run the following to invoke protoc via grpcio-tools:

pip install grpcio-tools
python -m grpc_tools.protoc -I . --python_betterproto_out=lib example.proto

This will generate lib/__init__.py which looks like:

# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: example.proto
# plugin: python-betterproto
from dataclasses import dataclass

import betterproto


@dataclass
class Greeting(betterproto.Message):
    """Greeting represents a message you can tell a user."""

    message: str = betterproto.string_field(1)

Then to use it:

>>> from lib import Greeting

>>> test = Greeting()
>>> test
Greeting(message='')

>>> test.message = "Hey!"
>>> test
Greeting(message="Hey!")

>>> bytes(test)
b'\n\x04Hey!'
>>> Greeting().parse(serialized)
Greeting(message="Hey!")

Async gRPC Support

The generated code includes grpclib based stub (client and server) classes for rpc services declared in the input proto files. It is enabled by default.

Given a service definition similar to the one below:

syntax = "proto3";

package echo;

message EchoRequest {
  string value = 1;
  // Number of extra times to echo
  uint32 extra_times = 2;
}

message EchoResponse {
  repeated string values = 1;
}

message EchoStreamResponse  {
  string value = 1;
}

service Echo {
  rpc Echo(EchoRequest) returns (EchoResponse);
  rpc EchoStream(EchoRequest) returns (stream EchoStreamResponse);
}

The generated client can be used like so:

import asyncio
from grpclib.client import Channel
import echo


async def main():
    channel = Channel(host="127.0.0.1", port=50051)
    service = echo.EchoStub(channel)
    response = await service.echo(value="hello", extra_times=1)
    print(response)

    async for response in service.echo_stream(value="hello", extra_times=1):
        print(response)

    # don't forget to close the channel when you're done!
    channel.close()

asyncio.run(main())  # python 3.7 only

# outputs
EchoResponse(values=['hello', 'hello'])
EchoStreamResponse(value='hello')
EchoStreamResponse(value='hello')

The server-facing stubs can be used to implement a Python gRPC server. To use them, simply subclass the base class in the generated files and override the service methods:

from echo import EchoBase
from grpclib.server import Server
from typing import AsyncIterator


class EchoService(EchoBase):
    async def echo(self, value: str, extra_times: int) -> "EchoResponse":
        return value

    async def echo_stream(
        self, value: str, extra_times: int
    ) -> AsyncIterator["EchoStreamResponse"]:
        for _ in range(extra_times):
            yield value


async def start_server():
    HOST = "127.0.0.1"
    PORT = 1337
    server = Server([EchoService()])
    await server.start(HOST, PORT)
    await server.serve_forever()

JSON

Message objects include betterproto.Message.to_json() and betterproto.Message.from_json() methods for JSON (de)serialisation, and betterproto.Message.to_dict(), betterproto.Message.from_dict() for converting back and forth from JSON serializable dicts.

For compatibility the default is to convert field names to betterproto.Casing.CAMEL. You can control this behavior by passing a different casing value, e.g:

@dataclass
class MyMessage(betterproto.Message):
    a_long_field_name: str = betterproto.string_field(1)


>>> test = MyMessage(a_long_field_name="Hello World!")
>>> test.to_dict(betterproto.Casing.SNAKE)
{"a_long_field_name": "Hello World!"}
>>> test.to_dict(betterproto.Casing.CAMEL)
{"aLongFieldName": "Hello World!"}

>>> test.to_json(indent=2)
'{\n  "aLongFieldName": "Hello World!"\n}'

>>> test.from_dict({"aLongFieldName": "Goodbye World!"})
>>> test.a_long_field_name
"Goodbye World!"