Betterproto2 Integration

Django-CFG supports betterproto2 for modern Python gRPC development with Pydantic dataclasses, native async support, and cleaner code generation.

Why Betterproto2?

Feature	Standard protobuf	Betterproto2
Message creation	`msg.field = value`	`Request(field=value)`
Type hints	Limited	Full dataclass support
Async support	Wrapper (grpcio)	Native (grpclib)
Serialization	Manual	Automatic
Pydantic integration	None	Built-in
IDE autocomplete	Partial	Full

Installation

Betterproto2 is included in the grpc extras:


pip install 'django-cfg[grpc]'

This installs:

betterproto2[all] - Runtime library with Pydantic support
grpclib - Native async gRPC library

For proto compilation, install dev dependencies:


pip install 'django-cfg[dev]'
# or
pip install betterproto2_compiler

Proto Compilation

Quick Compilation


from django_cfg.modules.django_grpc.utils import (
    compile_proto,
    compile_protos_directory,
)
 
# Compile single file
compile_proto("service.proto", "generated/")
 
# Compile directory
compile_protos_directory("protos/", "generated/", recursive=True)

Using BetterprotoCompiler

For more control:


from django_cfg.modules.django_grpc.utils.betterproto_compiler import (
    BetterprotoCompiler,
    BetterprotoCompilerConfig,
)
 
config = BetterprotoCompilerConfig(
    pydantic_dataclasses=True,
    include_google_types=True,
    generate_init=True,
)
 
compiler = BetterprotoCompiler(config)
compiler.compile_directory("protos/", "generated/")

Command Line


# Using protoc with betterproto2 plugin
python -m grpc_tools.protoc \
    -I protos/ \
    --python_betterproto2_out=generated/ \
    protos/*.proto

Use --python_betterproto2_out (with “2”), not --python_betterproto_out.

Generated Code Structure


generated/
├── __init__.py
├── myservice.py      # Messages and stubs
│   ├── MyRequest     # Pydantic dataclass
│   ├── MyResponse    # Pydantic dataclass
│   └── MyServiceStub # Async client stub

Betterproto2Client

The resilient client wrapper for betterproto2:

Basic Usage


from django_cfg.modules.django_grpc.services.client import Betterproto2Client
from generated import MyServiceStub, MyRequest
 
async with Betterproto2Client(host="localhost", port=50051) as client:
    stub = client.get_stub(MyServiceStub)
    response = await stub.my_method(MyRequest(field="value"))
    print(response)

With Resilience Options


async with Betterproto2Client(
    host="localhost",
    port=50051,
    enable_retry=True,
    enable_circuit_breaker=True,
    retry_attempts=5,
    circuit_breaker_threshold=5,
) as client:
    stub = client.get_stub(MyServiceStub)
 
    try:
        response = await stub.my_method(MyRequest(field="value"))
    except CircuitOpenError as e:
        print(f"Service unavailable, retry in {e.time_until_retry}s")

With Connection Pooling


async with Betterproto2Client(
    host="localhost",
    port=50051,
    use_pool=True,
) as client:
    stub = client.get_stub(MyServiceStub)
 
    # Multiple calls reuse the same channel
    for i in range(100):
        response = await stub.my_method(MyRequest(id=i))
 
    # Check pool stats
    print(client.pool_stats)

Using call_method

For resilience-wrapped calls:


async with Betterproto2Client(host="localhost", port=50051) as client:
    response = await client.call_method(
        stub_class=MyServiceStub,
        method_name="my_method",
        request=MyRequest(field="value"),
        request_id="req-001",  # For tracing
    )
 
    # Check circuit breaker status
    print(client.circuit_breaker_status)

Betterproto2ChannelPool

Dedicated channel pool for grpclib:


from django_cfg.modules.django_grpc.services.client import (
    Betterproto2ChannelPool,
    get_betterproto_pool,
    close_betterproto_pool,
)
 
# Global pool
pool = get_betterproto_pool()
 
async with pool.get_channel("localhost", 50051) as channel:
    stub = MyServiceStub(channel)
    response = await stub.my_method(request)
 
# Pool stats
print(pool.get_stats())
 
# Cleanup
await close_betterproto_pool()

ResilientStubWrapper

Add resilience to any existing stub:


from django_cfg.modules.django_grpc.services.client import ResilientStubWrapper
from grpclib.client import Channel
from generated import MyServiceStub, MyRequest
 
# Create channel and stub
channel = Channel(host="localhost", port=50051)
stub = MyServiceStub(channel)
 
# Wrap with resilience
resilient = ResilientStubWrapper(
    stub=stub,
    target_id="localhost:50051",
    enable_retry=True,
    enable_circuit_breaker=True,
)
 
# All calls now have retry and circuit breaker
response = await resilient.my_method(MyRequest(field="value"))
 
# Cleanup
channel.close()

Message Examples

Creating Messages


from generated import MyRequest, NestedData
 
# Simple message
request = MyRequest(
    id="123",
    name="test",
    active=True,
)
 
# With nested data
request = MyRequest(
    id="123",
    data=NestedData(
        field1="value1",
        field2=42,
    ),
)
 
# With repeated fields
request = MyRequest(
    id="123",
    tags=["tag1", "tag2", "tag3"],
)

Serialization


# To bytes
data = bytes(request)
 
# From bytes
request = MyRequest().parse(data)
 
# To dict
data = request.to_dict()
 
# From dict
request = MyRequest().from_dict(data)
 
# To JSON
json_str = request.to_json()
 
# From JSON
request = MyRequest().from_json(json_str)

Pydantic Validation


from pydantic import ValidationError
 
try:
    # Pydantic validates field types
    request = MyRequest(
        id=123,  # Error: expected str
        count="invalid",  # Error: expected int
    )
except ValidationError as e:
    print(e.errors())

Migration from Standard Protobuf

Step 1: Install Dependencies


pip install 'django-cfg[grpc]'
pip install betterproto2_compiler  # for compilation

Step 2: Regenerate Protos


from django_cfg.modules.django_grpc.utils import compile_protos_directory
 
# Generate to new directory
compile_protos_directory("protos/", "generated_bp2/")

Step 3: Update Imports


# Before (standard protobuf)
from generated.service_pb2 import MyRequest
from generated.service_pb2_grpc import MyServiceStub
 
# After (betterproto2)
from generated_bp2.service import MyRequest, MyServiceStub

Step 4: Update Message Creation


# Before
request = MyRequest()
request.id = "123"
request.name = "test"
request.data.field = "value"
 
# After
request = MyRequest(
    id="123",
    name="test",
    data=NestedData(field="value"),
)

Step 5: Update Client Code


# Before (grpcio)
async with grpc.aio.insecure_channel("localhost:50051") as channel:
    stub = MyServiceStub(channel)
    response = await stub.MyMethod(request)
 
# After (Betterproto2Client)
async with Betterproto2Client(host="localhost", port=50051) as client:
    stub = client.get_stub(MyServiceStub)
    response = await stub.my_method(request)

Note the method naming: betterproto2 uses snake_case (my_method) instead of PascalCase (MyMethod).

Configuration Reference

BetterprotoCompilerConfig


from django_cfg.modules.django_grpc.utils.betterproto_compiler import (
    BetterprotoCompilerConfig,
)
 
config = BetterprotoCompilerConfig(
    pydantic_dataclasses=True,   # Enable Pydantic validation
    include_google_types=True,    # Include well-known types
    generate_init=True,           # Generate __init__.py
    typing_imports=True,          # Add typing imports
    service_suffix="Stub",        # Suffix for service classes
)

Field	Type	Default	Description
`pydantic_dataclasses`	bool	True	Enable Pydantic
`include_google_types`	bool	True	Include google types
`generate_init`	bool	True	Generate init.py
`typing_imports`	bool	True	Add typing imports
`service_suffix`	str	”Stub”	Service class suffix

Checking Availability


from django_cfg.modules.django_grpc.services.client import HAS_GRPCLIB
 
if HAS_GRPCLIB:
    from django_cfg.modules.django_grpc.services.client import Betterproto2Client
    # Use betterproto2
else:
    from django_cfg.modules.django_grpc.services.client import AsyncResilientGRPCClient
    # Fallback to standard grpcio