Python Language Support

Status: ✅ Full Support
Example: examples/python-example/

Python support provides comprehensive Protocol Buffer integration with multiple output options including gRPC, type stubs, and modern dataclass alternatives.

Available Plugins

Plugin	Description	Generated Files
protoc-gen-python	Base message generation	*_pb2.py
protoc-gen-grpc_python	gRPC services	*_pb2_grpc.py
protoc-gen-pyi	Type stubs for IDE support	*_pb2.pyi
protoc-gen-mypy	Mypy type checking stubs	*_pb2.pyi, *_pb2_grpc.pyi
protoc-gen-python_betterproto	Modern dataclass approach	*.py (with @dataclass)
protoc-gen-python_betterproto (with pydantic)	Validated dataclasses	*.py (with Pydantic validation)

Configuration

Basic Configuration

languages.python = {
  enable = true;
  outputPath = "proto/gen/python";
};

Full Configuration

languages.python = {
  enable = true;
  outputPath = "proto/gen/python";
  options = [];

  # gRPC service generation
  grpc = {
    enable = true;
    options = [];
  };

  # Type stubs for better IDE support
  pyi = {
    enable = true;
    options = [];
  };

  # Mypy type checking support
  mypy = {
    enable = true;
    options = [];
  };

  # Modern Python dataclasses (alternative to standard protobuf)
  betterproto = {
    enable = false;  # Opt-in due to different API
    pydantic = false;  # Enable Pydantic dataclasses for validation
    options = [];
  };
};

Per-Language Files

🆕 New Feature

Control exactly which proto files Python processes using files and additionalFiles:

Data Science APIs

Use Case: Python needs data processing and ML proto files.

languages.python = {
  enable = true;
  outputPath = "ml_backend/proto";
  # Override: Only data science and ML-related protos
  files = [
    "./proto/common/v1/types.proto"
    "./proto/ml/v1/training_service.proto"
    "./proto/ml/v1/inference_service.proto"
    "./proto/data/v1/pipeline_service.proto"
    "./proto/analytics/v1/metrics.proto"
    # No frontend APIs - backend data processing only
  ];
  grpc.enable = true;
  mypy.enable = true;
};

Add External Dependencies

Use Case: Include third-party and Google Cloud proto definitions.

languages.python = {
  enable = true;
  # Use global files + additional Python-specific dependencies
  additionalFiles = [
    "./proto/google/cloud/storage/v1/storage.proto"
    "./proto/google/cloud/bigquery/v1/bigquery.proto"
    "./proto/third_party/tensorflow/serving/apis/model.proto"
    "./proto/monitoring/v1/prometheus.proto"
  ];
  grpc.enable = true;
  betterproto.enable = true;
};

Microservice Specific

Use Case: Python microservice only needs specific service definitions.

languages.python = {
  enable = true;
  outputPath = "auth_service/proto";
  # Override: Only auth and user management services
  files = [
    "./proto/common/v1/types.proto"
    "./proto/auth/v1/auth_service.proto"
    "./proto/users/v1/user_service.proto"
    "./proto/sessions/v1/session_service.proto"
    # No payment, analytics, or other services
  ];
  grpc.enable = true;
  pyi.enable = true;
};

Configuration Options

Option	Type	Description	Example
files	list	Override global files completely	Microservice-specific APIs
additionalFiles	list	Extend global files	Add Google Cloud protos

Common Python Use Cases

Data Science

Include ML training and inference service definitions

Cloud Integration

Add Google Cloud or AWS service proto definitions

Microservice Focus

Process only relevant service definitions per microservice

Third-Party APIs

Include external protobuf dependencies and frameworks

Proto Example

proto/example/v1/example.proto

syntax = "proto3";

package example.v1;

message Greeting {
  string id = 1;
  string content = 2;
  int64 created_at = 3;
  repeated string tags = 4;
  map<string, string> metadata = 5;
  oneof optional_field {
    string text_value = 6;
    int32 numeric_value = 7;
  }
}

enum GreetingType {
  GREETING_TYPE_UNSPECIFIED = 0;
  GREETING_TYPE_HELLO = 1;
  GREETING_TYPE_GOODBYE = 2;
  GREETING_TYPE_WELCOME = 3;
}

message CreateGreetingRequest {
  string content = 1;
  repeated string tags = 2;
  GreetingType type = 3;
}

message CreateGreetingResponse {
  Greeting greeting = 1;
}

message ListGreetingsRequest {
  int32 page_size = 1;
  string page_token = 2;
  GreetingType type_filter = 3;
}

message ListGreetingsResponse {
  repeated Greeting greetings = 1;
  string next_page_token = 2;
}

service GreetingService {
  rpc CreateGreeting(CreateGreetingRequest) returns (CreateGreetingResponse);
  rpc ListGreetings(ListGreetingsRequest) returns (ListGreetingsResponse);
  rpc StreamGreetings(ListGreetingsRequest) returns (stream Greeting);
  rpc BidirectionalStream(stream CreateGreetingRequest) returns (stream Greeting);
}

Generated Code Usage

Standard Protobuf

import grpc
from proto.gen.python.example.v1 import example_pb2
from proto.gen.python.example.v1 import example_pb2_grpc

# Client usage
async def main():
    # Create channel and stub
    channel = grpc.insecure_channel('localhost:50051')
    stub = example_pb2_grpc.GreetingServiceStub(channel)

    # Create a greeting
    request = example_pb2.CreateGreetingRequest(
        content="Hello from Python!",
        tags=["python", "grpc", "example"],
        type=example_pb2.GREETING_TYPE_HELLO
    )

    response = stub.CreateGreeting(request)
    print(f"Created greeting: {response.greeting.id}")

    # Work with the message
    greeting = response.greeting
    print(f"Content: {greeting.content}")
    print(f"Tags: {list(greeting.tags)}")

    # Access map fields
    greeting.metadata["author"] = "Python Client"
    greeting.metadata["version"] = "1.0"

    # Serialize to bytes
    data = greeting.SerializeToString()
    print(f"Serialized size: {len(data)} bytes")

    # Deserialize from bytes
    new_greeting = example_pb2.Greeting()
    new_greeting.ParseFromString(data)

    # JSON support
    from google.protobuf.json_format import MessageToJson, Parse
    json_str = MessageToJson(greeting)
    print(f"JSON: {json_str}")

    # Parse from JSON
    from_json = Parse(json_str, example_pb2.Greeting())

Async gRPC Server

import grpc
import asyncio
from concurrent import futures
import time
from proto.gen.python.example.v1 import example_pb2
from proto.gen.python.example.v1 import example_pb2_grpc

class GreetingServicer(example_pb2_grpc.GreetingServiceServicer):
    def __init__(self):
        self.greetings = {}
        self.counter = 0

    def CreateGreeting(self, request, context):
        self.counter += 1
        greeting = example_pb2.Greeting(
            id=f"greeting-{self.counter}",
            content=request.content,
            created_at=int(time.time())
        )
        greeting.tags.extend(request.tags)

        # Set optional field based on type
        if request.type == example_pb2.GREETING_TYPE_HELLO:
            greeting.text_value = "Hello variant"
        else:
            greeting.numeric_value = request.type

        self.greetings[greeting.id] = greeting

        return example_pb2.CreateGreetingResponse(greeting=greeting)

    def ListGreetings(self, request, context):
        filtered = [
            g for g in self.greetings.values()
            if request.type_filter == 0 or
               self._get_type(g) == request.type_filter
        ]

        # Simple pagination
        start = 0 if not request.page_token else int(request.page_token)
        end = start + request.page_size

        page = filtered[start:end]
        next_token = str(end) if end < len(filtered) else ""

        return example_pb2.ListGreetingsResponse(
            greetings=page,
            next_page_token=next_token
        )

    def StreamGreetings(self, request, context):
        """Server streaming RPC"""
        while context.is_active():
            for greeting in self.greetings.values():
                if request.type_filter == 0 or \
                   self._get_type(greeting) == request.type_filter:
                    yield greeting
                    time.sleep(1)

    def BidirectionalStream(self, request_iterator, context):
        """Bidirectional streaming RPC"""
        for request in request_iterator:
            # Process each request and yield response
            response = self.CreateGreeting(request, context)
            yield response.greeting

    def _get_type(self, greeting):
        # Helper to determine greeting type
        if greeting.HasField("text_value"):
            return example_pb2.GREETING_TYPE_HELLO
        elif greeting.HasField("numeric_value"):
            return greeting.numeric_value
        return example_pb2.GREETING_TYPE_UNSPECIFIED

# Start server
async def serve():
    server = grpc.aio.server()
    example_pb2_grpc.add_GreetingServiceServicer_to_server(
        GreetingServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    await server.start()
    print("Server started on port 50051")
    await server.wait_for_termination()

if __name__ == '__main__':
    asyncio.run(serve())

Betterproto

# When betterproto is enabled, you get modern dataclasses
import asyncio
import grpclib
from proto.gen.python.example.v1 import Greeting, CreateGreetingRequest
from proto.gen.python.example.v1 import GreetingServiceStub

async def main():
    # Betterproto uses dataclasses
    greeting = Greeting(
        id="greeting-1",
        content="Hello from betterproto!",
        created_at=int(time.time()),
        tags=["python", "betterproto"],
        metadata={"author": "Python Client"},
        text_value="Hello variant"  # Oneof fields are normal attributes
    )

    # Clean pythonic API
    print(f"Greeting: {greeting}")
    print(f"Has text value: {greeting.text_value is not None}")

    # Async/await native support
    async with grpclib.client.Channel('localhost', 50051) as channel:
        client = GreetingServiceStub(channel)

        # Create greeting
        request = CreateGreetingRequest(
            content="Hello betterproto!",
            tags=["modern", "async"]
        )
        response = await client.create_greeting(request)

        # Stream greetings
        async for greeting in client.stream_greetings(
            ListGreetingsRequest(page_size=10)
        ):
            print(f"Streamed: {greeting.content}")

    # Serialization
    data = bytes(greeting)  # Clean serialization
    new_greeting = Greeting().parse(data)  # Clean parsing

    # JSON support
    json_str = greeting.to_json()
    from_json = Greeting().from_json(json_str)

Betterproto + Pydantic

# With pydantic enabled, you get validation in addition to dataclasses
import asyncio
import grpclib
from pydantic import ValidationError
from proto.gen.python.example.v1 import Greeting, CreateGreetingRequest
from proto.gen.python.example.v1 import GreetingServiceStub

async def main():
    # Pydantic provides validation on creation
    try:
        greeting = Greeting(
            id="greeting-1",
            content="Hello from Pydantic betterproto!",
            created_at=int(time.time()),
            tags=["python", "pydantic", "validation"],
            metadata={"author": "Python Client"},
            text_value="Hello variant"
        )
        print(f"Valid greeting created: {greeting}")
    except ValidationError as e:
        print(f"Validation failed: {e}")

    # Type coercion works automatically
    validated_greeting = Greeting(
        id="auto-coerced",
        content="Test",
        created_at="1234567890",  # String will be coerced to int
        price="99.99",  # String will be coerced to float
        tags=["test"]
    )
    print(f"Coerced types: created_at={type(validated_greeting.created_at)}")

    # Enhanced validation (with proper proto field constraints)
    try:
        # This would validate if field constraints were defined in proto
        invalid_greeting = Greeting(
            id="invalid",
            content="",  # Empty content might be invalid
            created_at=-1,  # Negative timestamp might be invalid
            price=-10.0  # Negative price might be invalid
        )
    except ValidationError as e:
        print(f"Validation caught invalid data: {e}")

    # Pydantic methods available
    if hasattr(greeting, 'model_dump'):
        data_dict = greeting.model_dump()
        print(f"Model dump: {data_dict}")

    if hasattr(greeting, 'model_validate'):
        # Validate external data
        external_data = {
            "id": "external-1",
            "content": "From external source",
            "created_at": 1234567890,
            "tags": ["external"],
            "metadata": {}
        }
        validated = Greeting.model_validate(external_data)
        print(f"Validated external data: {validated}")

    # Async/await support still works
    async with grpclib.client.Channel('localhost', 50051) as channel:
        client = GreetingServiceStub(channel)

        request = CreateGreetingRequest(
            content="Pydantic validation enabled!",
            tags=["pydantic", "validated"]
        )
        response = await client.create_greeting(request)
        print(f"Server response: {response}")

    # Enhanced error messages
    try:
        Greeting(
            id=None,  # Invalid type
            content="Test"
        )
    except ValidationError as e:
        print(f"Enhanced error message: {e}")

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

Type Hints

# With mypy/pyi enabled, you get full type hints
from typing import Optional, List, Dict
from proto.gen.python.example.v1 import example_pb2

def process_greeting(greeting: example_pb2.Greeting) -> Dict[str, any]:
    """Process a greeting with full type safety"""
    result: Dict[str, any] = {
        "id": greeting.id,
        "content": greeting.content,
        "tags": list(greeting.tags),
        "metadata": dict(greeting.metadata),
        "created_at": greeting.created_at
    }

    # Type checker knows about oneof fields
    if greeting.HasField("text_value"):
        result["text"] = greeting.text_value
    elif greeting.HasField("numeric_value"):
        result["number"] = greeting.numeric_value

    return result

def create_greeting(
    content: str,
    tags: Optional[List[str]] = None,
    metadata: Optional[Dict[str, str]] = None
) -> example_pb2.Greeting:
    """Create a greeting with optional parameters"""
    greeting = example_pb2.Greeting(
        id=f"greeting-{int(time.time())}",
        content=content,
        created_at=int(time.time())
    )

    if tags:
        greeting.tags.extend(tags)

    if metadata:
        greeting.metadata.update(metadata)

    return greeting

# IDE will provide autocompletion and type checking
greeting = create_greeting(
    "Hello",
    tags=["python", "typed"],
    metadata={"author": "Type Safe"}
)

Testing

import pytest
from proto.gen.python.example.v1 import example_pb2

def test_greeting_creation():
    """Test creating and serializing a greeting"""
    greeting = example_pb2.Greeting(
        id="test-1",
        content="Test greeting",
        created_at=1234567890
    )
    greeting.tags.extend(["test", "unit"])
    greeting.metadata["test"] = "true"

    # Test serialization
    data = greeting.SerializeToString()
    assert len(data) > 0

    # Test deserialization
    new_greeting = example_pb2.Greeting()
    new_greeting.ParseFromString(data)

    assert new_greeting.id == "test-1"
    assert new_greeting.content == "Test greeting"
    assert list(new_greeting.tags) == ["test", "unit"]
    assert dict(new_greeting.metadata) == {"test": "true"}

@pytest.mark.asyncio
async def test_grpc_service():
    """Test gRPC service calls"""
    # Mock or integration test your service
    pass

Best Practices

1. Use Type Stubs: Enable pyi or mypy for better IDE support
2. Async Support: Use grpc.aio for async/await support
3. Error Handling: Always handle grpc.RpcError exceptions
4. Streaming: Use generators efficiently for streaming RPCs
5. Betterproto: Consider for new projects wanting modern Python APIs
6. Pydantic Validation: Enable betterproto.pydantic = true for data validation
7. Package Structure: Follow Python package conventions
8. Validation Strategy: Use proto field constraints for comprehensive validation

Package Structure

• Directoryproto/ - gen/ - python/ - example/ - v1/ - __init__.py - example_pb2.py

  • …
• example_pb2.pyi If pyi/mypy enabled - example_pb2_grpc.py - example_pb2_grpc.pyi If pyi/mypy enabled

Performance Considerations

• Standard protobuf: Best performance, C++ implementation
• Betterproto: Pure Python, slower but cleaner API
• Betterproto + Pydantic: Additional validation overhead, enhanced safety
• Serialization: Binary format is much smaller than JSON
• Streaming: More efficient for large datasets
• Validation: Pydantic validation has runtime cost but prevents data errors

Try the Example

# Standard Python example
cd examples/python-example
nix develop
bufrnix_init
bufrnix
python main.py
pytest -v

# Betterproto with Pydantic validation
cd examples/python-betterproto-pydantic
nix develop
nix run  # Generate code
python test_pydantic.py  # Test validation features

Troubleshooting

Import Errors

Ensure the generated code directory is in your Python path:

import sys
sys.path.append('./proto/gen/python')

Type Checking

For mypy to work correctly, use:

mypy --follow-imports=skip proto/gen/python/

Betterproto Compatibility

Betterproto generates different code than standard protobuf. Don’t mix them in the same project.

Pydantic Validation

For comprehensive validation with Pydantic:

1. Define field constraints in your .proto files
2. Use proto validation annotations (e.g., buf validate)
3. Handle ValidationError exceptions in your code
4. Consider validation performance impact for high-throughput scenarios

Try the Example

{
  description = "Python with betterproto example";

  inputs = {
    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
    flake-utils.url = "github:numtide/flake-utils";
    bufrnix = {
      url = "github:conneroisu/bufrnix";
      inputs.nixpkgs.follows = "nixpkgs";
    };
  };

  outputs = {
    self,
    nixpkgs,
    flake-utils,
    bufrnix,
  }:
    flake-utils.lib.eachDefaultSystem (system: let
      pkgs = nixpkgs.legacyPackages.${system};

      bufrnixConfig = bufrnix.lib.mkBufrnix {
        inherit pkgs;
        config = {
          root = "./proto";

          # Python with betterproto for modern dataclasses
          languages.python = {
            enable = true;
            outputPath = "proto/gen/python";

            # Per-language file control (new feature)
            # files = [
            #   "./proto/common/v1/types.proto"
            #   "./proto/ml/v1/training_service.proto"
            # ];
            # additionalFiles = [
            #   "./proto/google/cloud/storage/v1/storage.proto"
            #   "./proto/third_party/tensorflow/serving/apis/model.proto"
            # ];

            # Use betterproto instead of standard protobuf
            betterproto = {
              enable = true;
            };
          };
        };
      };
    in {
      devShells.default = pkgs.mkShell {
        buildInputs = with pkgs; [
          python3
          python3Packages.betterproto
          python3Packages.grpclib # betterproto uses grpclib instead of grpcio
          protobuf
        ];

        shellHook = ''
          echo "Python Betterproto Example"
          echo "========================="
          echo "Commands:"
          echo "  bufrnix_init - Initialize project"
          echo "  bufrnix - Generate betterproto code"
          echo "  python test_betterproto.py - Run test"
          echo ""
          echo "Note: Betterproto generates modern Python dataclasses"
          echo "      with async support and cleaner API"
          echo ""
          ${bufrnixConfig.shellHook}
        '';
      };
    });
}