A better Protobuf / gRPC generator & library
npx @tessl/cli install tessl/pypi-betterproto@1.2.0A modern, idiomatic Python implementation of Protocol Buffers (protobuf) and gRPC code generation that addresses limitations of the official Google protoc plugin. It generates clean, readable Python code using modern language features including dataclasses, async/await, type hints, and timezone-aware datetime objects for Python 3.6+.
pip install betterproto (runtime) or pip install "betterproto[compiler]" (with code generation)import betterprotoFor message classes and field definitions:
from betterproto import Message, EnumFor field creation functions:
from betterproto import (
string_field, int32_field, bool_field, message_field,
enum_field, bytes_field, map_field
)from dataclasses import dataclass
from typing import List
import betterproto
@dataclass
class Person(betterproto.Message):
name: str = betterproto.string_field(1)
age: int = betterproto.int32_field(2)
email: str = betterproto.string_field(3)
@dataclass
class AddressBook(betterproto.Message):
people: List[Person] = betterproto.message_field(1)# Create a message
person = Person(name="Alice", age=30, email="alice@example.com")
# Serialize to binary
binary_data = bytes(person)
# Parse from binary
parsed_person = Person().parse(binary_data)
# JSON serialization
json_str = person.to_json()
person_from_json = Person().from_json(json_str)
# Dictionary conversion
person_dict = person.to_dict()
person_from_dict = Person().from_dict(person_dict)import asyncio
from grpclib.client import Channel
import betterproto
class GreeterServiceStub(betterproto.ServiceStub):
async def say_hello(
self,
request: HelloRequest,
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional[betterproto._MetadataLike] = None,
) -> HelloReply:
return await self._unary_unary(
"/helloworld.Greeter/SayHello",
request,
HelloReply,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
async def main():
async with Channel("localhost", 50051) as channel:
greeter = GreeterServiceStub(channel)
reply = await greeter.say_hello(HelloRequest(name="World"))
print(f"Greeting: {reply.message}")BetterProto's architecture consists of several key components:
betterproto.Message providing serialization, parsing, and JSON conversionFieldMetadata dataclass stores protobuf-specific information (field numbers, types, wire formats)betterproto.plugin generates Python dataclasses from .proto filesServiceStub base class for async gRPC client generation with grpclib supportCore message functionality including field creation functions for all protobuf types, dataclass integration, and metadata handling for proper serialization and deserialization.
class Message:
def __bytes__(self) -> bytes: ...
def parse(self, data: bytes) -> T: ...
def to_dict(self, casing: Casing = Casing.CAMEL, include_default_values: bool = False) -> dict: ...
def from_dict(self, value: dict) -> T: ...
def to_json(self, indent: Union[None, int, str] = None) -> str: ...
def from_json(self, value: Union[str, bytes]) -> T: ...
def string_field(number: int, group: Optional[str] = None) -> Any: ...
def int32_field(number: int, group: Optional[str] = None) -> Any: ...
def int64_field(number: int, group: Optional[str] = None) -> Any: ...
def bool_field(number: int, group: Optional[str] = None) -> Any: ...
def bytes_field(number: int, group: Optional[str] = None) -> Any: ...
def message_field(number: int, group: Optional[str] = None, wraps: Optional[str] = None) -> Any: ...Protobuf enumeration support with integer-based enums that provide string name conversion and integration with the message system.
class Enum(int, enum.Enum):
@classmethod
def from_string(cls, name: str) -> int: ...
def enum_field(number: int, group: Optional[str] = None) -> Any: ...Async gRPC client stub generation with support for unary and streaming calls, timeout handling, metadata, and deadline management using grpclib.
class ServiceStub:
def __init__(
self,
channel: Channel,
*,
timeout: Optional[float] = None,
deadline: Optional[Deadline] = None,
metadata: Optional[_MetadataLike] = None,
) -> None: ...
async def _unary_unary(
self,
route: str,
request: IProtoMessage,
response_type: Type[T],
*,
timeout: Optional[float] = None,
deadline: Optional[Deadline] = None,
metadata: Optional[_MetadataLike] = None,
) -> T: ...Protocol buffer compiler plugin that generates clean Python dataclasses from .proto files with proper type hints, async gRPC stubs, and modern Python conventions.
def main() -> None: ...
def generate_code(request, response) -> None: ...Low-level serialization utilities including varint encoding/decoding, wire type handling, and binary format parsing compatible with standard protobuf implementations.
def encode_varint(value: int) -> bytes: ...
def decode_varint(buffer: bytes, pos: int, signed: bool = False) -> Tuple[int, int]: ...
def parse_fields(value: bytes) -> Generator[ParsedField, None, None]: ...
def serialized_on_wire(message: Message) -> bool: ...Helper functions for message introspection, one-of field handling, and casing conversion to support generated code and user applications.
def which_one_of(message: Message, group_name: str) -> Tuple[str, Any]: ...
def safe_snake_case(value: str) -> str: ...@dataclass(frozen=True)
class FieldMetadata:
number: int
proto_type: str
map_types: Optional[Tuple[str, str]] = None
group: Optional[str] = None
wraps: Optional[str] = None
@dataclass(frozen=True)
class ParsedField:
number: int
wire_type: int
value: Any
raw: bytes
class Casing(enum.Enum):
CAMEL: Callable
SNAKE: Callable
# Type variables and aliases
T = TypeVar('T', bound='Message') # Bound type variable for Message subclasses
_Value = Union[str, bytes]
_MetadataLike = Union[Mapping[str, _Value], Collection[Tuple[str, _Value]]]