A better Protobuf / gRPC generator & library
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Protocol buffer compiler plugin that generates clean Python dataclasses from .proto files with proper type hints, async gRPC stubs, and modern Python conventions.
The main entry point function for the protoc plugin that processes protobuf definitions and generates Python code.
def main() -> None:
"""
The plugin's main entry point.
Reads CodeGeneratorRequest from stdin, processes protobuf definitions,
and writes CodeGeneratorResponse to stdout.
"""The core function that transforms protobuf definitions into Python code.
def generate_code(request, response) -> None:
"""
Generate Python code from protobuf definitions.
Args:
request: CodeGeneratorRequest from protoc
response: CodeGeneratorResponse to populate with generated files
"""Functions that convert protobuf types to appropriate Python types and handle type references.
def get_ref_type(
package: str,
imports: set,
type_name: str,
unwrap: bool = True
) -> str:
"""
Return a Python type name for a proto type reference.
Args:
package: Current package name
imports: Set to add import statements to
type_name: Protobuf type name
unwrap: Whether to unwrap Google wrapper types
Returns:
Python type string
"""
def py_type(
package: str,
imports: set,
message: DescriptorProto,
descriptor: FieldDescriptorProto,
) -> str:
"""
Get Python type for a protobuf field descriptor.
Args:
package: Current package name
imports: Set to add import statements to
message: Message containing the field
descriptor: Field descriptor
Returns:
Python type string for the field
"""
def get_py_zero(type_num: int) -> str:
"""
Get Python zero/default value for a protobuf type number.
Args:
type_num: Protobuf type number
Returns:
String representation of the default value
"""Functions for navigating and extracting information from protobuf definitions.
def traverse(proto_file):
"""
Traverse protobuf file structure yielding items and their paths.
Args:
proto_file: Protobuf file descriptor
Yields:
Tuples of (item, path) for messages and enums
"""
def get_comment(proto_file, path: List[int], indent: int = 4) -> str:
"""
Extract comments from protobuf source code info.
Args:
proto_file: Protobuf file descriptor
path: Path to the element in the descriptor
indent: Indentation level for the comment
Returns:
Formatted comment string
"""# Install betterproto with compiler support
pip install "betterproto[compiler]"
# Generate Python code from .proto files
protoc -I . --python_betterproto_out=. example.proto
# Generate code for multiple files
protoc -I . --python_betterproto_out=. *.proto
# Specify include paths
protoc -I ./protos -I ./common --python_betterproto_out=./generated example.protosyntax = "proto3";
package example;
// A simple greeting message
message HelloRequest {
string name = 1; // The name to greet
int32 count = 2; // Number of greetings
}
// Response with greeting
message HelloReply {
string message = 1;
repeated string additional_messages = 2;
}
// Greeting service
service Greeter {
// Sends a greeting
rpc SayHello (HelloRequest) returns (HelloReply);
// Sends multiple greetings
rpc SayManyHellos (HelloRequest) returns (stream HelloReply);
}The plugin generates clean, typed Python code:
# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: example.proto
# plugin: python-betterproto
from dataclasses import dataclass
from typing import AsyncGenerator, List, Optional
import betterproto
from grpclib.client import Channel
@dataclass
class HelloRequest(betterproto.Message):
"""A simple greeting message"""
# The name to greet
name: str = betterproto.string_field(1)
# Number of greetings
count: int = betterproto.int32_field(2)
@dataclass
class HelloReply(betterproto.Message):
"""Response with greeting"""
message: str = betterproto.string_field(1)
additional_messages: List[str] = betterproto.string_field(2)
class GreeterStub(betterproto.ServiceStub):
"""Greeting service"""
async def say_hello(
self,
request: HelloRequest,
*,
timeout: Optional[float] = None,
deadline: Optional["betterproto.Deadline"] = None,
metadata: Optional["betterproto._MetadataLike"] = None,
) -> HelloReply:
"""Sends a greeting"""
return await self._unary_unary(
"/example.Greeter/SayHello",
request,
HelloReply,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
async def say_many_hellos(
self,
request: HelloRequest,
*,
timeout: Optional[float] = None,
deadline: Optional["betterproto.Deadline"] = None,
metadata: Optional["betterproto._MetadataLike"] = None,
) -> AsyncGenerator[HelloReply, None]:
"""Sends multiple greetings"""
async for response in self._unary_stream(
"/example.Greeter/SayManyHellos",
request,
HelloReply,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield responseThe plugin handles advanced protobuf features:
syntax = "proto3";
package advanced;
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
enum Status {
UNKNOWN = 0;
ACTIVE = 1;
INACTIVE = 2;
}
message User {
string id = 1;
string name = 2;
Status status = 3;
// One-of fields
oneof contact {
string email = 4;
string phone = 5;
}
// Repeated and nested
repeated Address addresses = 6;
// Map fields
map<string, string> metadata = 7;
// Google types
google.protobuf.Timestamp created_at = 8;
google.protobuf.StringValue nickname = 9;
}
message Address {
string street = 1;
string city = 2;
string country = 3;
}Generated code with proper typing:
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional
import betterproto
class Status(betterproto.Enum):
UNKNOWN = 0
ACTIVE = 1
INACTIVE = 2
@dataclass
class Address(betterproto.Message):
street: str = betterproto.string_field(1)
city: str = betterproto.string_field(2)
country: str = betterproto.string_field(3)
@dataclass
class User(betterproto.Message):
id: str = betterproto.string_field(1)
name: str = betterproto.string_field(2)
status: Status = betterproto.enum_field(3)
# One-of fields
email: str = betterproto.string_field(4, group="contact")
phone: str = betterproto.string_field(5, group="contact")
# Repeated and nested
addresses: List[Address] = betterproto.message_field(6)
# Map fields
metadata: Dict[str, str] = betterproto.map_field(7, "string", "string")
# Google types (automatically unwrapped)
created_at: datetime = betterproto.message_field(8)
nickname: Optional[str] = betterproto.message_field(9, wraps="string")The plugin can be configured through command-line options:
# Basic usage
protoc --python_betterproto_out=. example.proto
# With custom output directory
protoc --python_betterproto_out=./output example.proto
# Multiple include paths
protoc -I./protos -I./vendor --python_betterproto_out=. example.proto# setup.py
from setuptools import setup
from betterproto.plugin import generate_proto_code
setup(
name="my-package",
# ... other setup parameters
)
# Custom command to generate proto code
if __name__ == "__main__":
# Generate proto code before building
import subprocess
subprocess.run([
"protoc",
"-I", "protos",
"--python_betterproto_out=src/generated",
"protos/*.proto"
])# Makefile
.PHONY: generate-proto
generate-proto:
protoc -I protos --python_betterproto_out=src/generated protos/*.proto
.PHONY: clean-proto
clean-proto:
rm -rf src/generated/*.py
build: generate-proto
python -m build# Google wrapper type mappings
WRAPPER_TYPES: Dict[str, Optional[Type]] = {
"google.protobuf.DoubleValue": google_wrappers.DoubleValue,
"google.protobuf.FloatValue": google_wrappers.FloatValue,
"google.protobuf.Int64Value": google_wrappers.Int64Value,
"google.protobuf.UInt64Value": google_wrappers.UInt64Value,
"google.protobuf.Int32Value": google_wrappers.Int32Value,
"google.protobuf.UInt32Value": google_wrappers.UInt32Value,
"google.protobuf.BoolValue": google_wrappers.BoolValue,
"google.protobuf.StringValue": google_wrappers.StringValue,
"google.protobuf.BytesValue": google_wrappers.BytesValue,
}Install with Tessl CLI
npx tessl i tessl/pypi-betterproto