Protocol Buffers (protobuf) relies on well-defined data types to build efficient data structures. Understanding these types and how to write .proto files is foundational for using gRPC. This article introduces common data types and the structure and writing conventions of .proto files, then walks through implementing Python gRPC services with a focus on streaming and a firmware upgrade case study.

Protobuf Data Types

Protobuf supports multiple data types to meet various business needs. They fall into the following categories:

Scalar Types

These are the most basic data types, mapping to fundamental types in programming languages.

Protobuf Type	Description	Python Equivalent
double / float	64-bit / 32-bit floating point	float
int32 / int64	Variable-length signed integers. For negative numbers, sintN is more efficient.	int
uint32 / uint64	Variable-length unsigned integers.	int
sint32 / sint64	Variable-length signed integers. More efficient than int32/int64 for negative values.	int
fixed32 / fixed64	Fixed 4-byte / 8-byte unsigned integers. More efficient than uint32/uint64 when values often exceed 2^28.	int
sfixed32 / sfixed64	Fixed 4-byte / 8-byte signed integers.	int
bool	Boolean value	bool
string	UTF-8 encoded or 7-bit ASCII text.	str (Unicode)
bytes	Arbitrary byte sequence.	bytes

Composite Types

These types enable building more complex data structures.

• Enumeration (enum): Defines a set of predefined constant values.

  enum DeviceStatus {
    DEVICE_STATUS_UNKNOWN = 0; // Must start from 0
    DEVICE_STATUS_ONLINE = 1;
    DEVICE_STATUS_OFFLINE = 2;
  }
  

• Message Type (message): User-defined complex data types that can contain any other types (including other message types), analogous to classes.

  message FirmwareInfo {
    string product_id = 1;
    string version = 2;
    int64 size_bytes = 3;
  }
  

• Nested Types: Define a message inside another message to organize closely related data.

Special Types

• Repeated Fields (repeated): Indicates the field can contain zero or more values, similar to an array or list.

  message UploadResult {
    repeated string error_messages = 1;
  }
  

• Map Type (map): Defines a collection of key-value pairs, similar to a dictionary.

  message UpdateReport {
    map<string, string> metadata = 1;
  }
  

• Opsional Fields and oneof:

  • In proto3 syntax, all fields are "optional"; unset fields return a default value (empty string, 0, false).
  • oneof: Indicates that at most one of the contained fields can be set at a time, useful for mutually exclusive options.

    message UpdateRequest {
      oneof payload {
        string manifest_url = 1;
        bytes firmware_image = 2;
      }
    }
    

.proto File Writing Conventions

A well-structured .proto file typically contains the following sections:

1. Syntax Declaration

syntax = "proto3";

2. Package Declaration (Optional)

Prevents naming conflicts and acts as a namespace in generated code.

package firmware_update;

3. Imports (Optional)

Reuse types defined in other .proto files.

import "google/protobuf/timestamp.proto";

4. Message Definitions (message)

Define data structures using the available types.

5. Field Numbers

Every field in a message must have a unique number. These numbers identify fields in the binary encoding and must not be changed after deployment. Numbers 1 to 536,870,911 are valid (19000 to 19999 are reserved). Assign numbers 1 to 15 to the most frequent fields for a more compact encoding (1 byte vs. 2 bytes for numbers up to 2047).

6. Service Definitions (service)

Define RPC interfaces for gRPC, specifying method names, request parameters, and return types.

service FirmwareService {
  rpc Upload(stream FirmwareChunk) returns (UploadStatus);
  rpc GetVersion(VersionRequest) returns (VersionInfo);
}

7. Comments

Use // or /* ... */ for documentation.

8. Reserved Fields (reserved)

When deleting a field, add its number (and name) to a reserved list to prevent future reuse, which could cause data corruption.

message DeprecatedMessage {
  reserved 2, 15, 9 to 11;
  reserved "old_field_name";
}

Python gRPC Tutorial

Environment Setup

Install the required packages:

pip install grpcio
pip install grpcio-tools

Verify with the Helloworld example:

git clone -b v1.74.0 --depth 1 --shallow-submodules https://github.com/grpc/grpc
cd grpc/examples/python/helloworld
python greeter_server.py
# In another terminal
python greeter_client.py

Core Concepts: Protocol Buffers and Service Definition

A typical .proto file contains messages and services:

syntax = "proto3";

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

Four Service Method Types

Type	Syntax	Description	Use Case
Unary RPC	rpc Method(Request) returns (Response)	Client sends one request, server returns one response.	Simple queries, status retrieval (GetSWInfos).
Server Streaming RPC	rpc Method(Request) returns (stream Response)	Client sends one request, server returns a stream of responses.	Pushing large data sets, log downloads.
Client Streaming RPC	rpc Method(stream Request) returns (Response)	Client sends a stream of requests, server returns a single response.	Large file uploads (e.g., firmware upgrade), data collection.
Bidirectional Streaming RPC	rpc Method(stream Request) returns (stream Response)	Both sides send/receive messages independently via streams.	Real-time chat, game state sync.

Generating Python Code from .proto

python -m grpc_tools.protoc \
    -I../../protos \
    --python_out=. \
    --grpc_python_out=. \
    ../../protos/route_guide.proto

This generates *_pb2.py (message classes) and *_pb2_grpc.py (service stubs and servicer base classes).

Implementing a Python gRPC Server

Create a subclass of the generated Servicer class and implement the RPC methods.

import route_guide_pb2
import route_guide_pb2_grpc

class RouteGuideServicer(route_guide_pb2_grpc.RouteGuideServicer):
    def GetFeature(self, request, context):
        feature = get_feature_from_db(request.latitude, request.longitude)
        if feature is None:
            return route_guide_pb2.Feature(name="", location=request)
        return feature

    def RecordRoute(self, request_iterator, context):
        point_count = 0
        for point in request_iterator:
            point_count += 1
        return route_guide_pb2.RouteSummary(point_count=point_count)

    def RouteChat(self, request_iterator, context):
        for note in request_iterator:
            yield route_guide_pb2.RouteNote(message=f"Echo: {note.message}", location=note.location)

Start the server:

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    route_guide_pb2_grpc.add_RouteGuideServicer_to_server(RouteGuideServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

Implementing a Python gRPC Client

Create a channel and stub, then call methods. For streaming RPCs, use generator functions.

import grpc
import route_guide_pb2
import route_guide_pb2_grpc

channel = grpc.insecure_channel('localhost:50051')
stub = route_guide_pb2_grpc.RouteGuideStub(channel)

# Unary
point = route_guide_pb2.Point(latitude=409146138, longitude=-746188906)
feature = stub.GetFeature(point)

# Client Streaming
def generate_points():
    points = [
        route_guide_pb2.Point(latitude=407838351, longitude=-746143763),
        route_guide_pb2.Point(latitude=408122808, longitude=-743999179),
    ]
    for pt in points:
        yield pt

summary = stub.RecordRoute(generate_points())

# Bidirectional Streaming
def generate_notes():
    notes = [
        route_guide_pb2.RouteNote(message="First", location=point1),
        route_guide_pb2.RouteNote(message="Second", location=point2),
    ]
    for note in notes:
        yield note

responses = stub.RouteChat(generate_notes())
for response in responses:
    print(f"Received: {response.message}")

Four Service Methods in a Firmware Upgrade System

We define a unified .proto file for a firmware upgrade service to demonstrate all four types.

syntax = "proto3";

package firmware;

service FirmwareService {
  rpc GetVersion(VersionRequest) returns (VersionInfo);
  rpc DownloadLogs(LogRequest) returns (stream LogEntry);
  rpc UploadFirmware(stream FirmwareChunk) returns (UploadStatus);
  rpc MonitorUpgrade(stream ControlCommand) returns (stream ProgressReport);
}

message VersionRequest {}
message VersionInfo {
  string version = 1;
  string build_time = 2;
}

message LogRequest {
  int32 max_lines = 1;
}
message LogEntry {
  string line = 1;
  int64 timestamp = 2;
}

message FirmwareChunk {
  bytes data = 1;
  string product_id = 2;
}
message UploadStatus {
  int32 code = 1;
  string message = 2;
}

message ControlCommand {
  enum Command {
    START = 0;
    PAUSE = 1;
    RESUME = 2;
    CANCEL = 3;
  }
  Command cmd = 1;
}
message ProgressReport {
  int32 percent = 1;
  string stage = 2;
  string message = 3;
}

Generate the Python code:

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. firmware.proto

Unary RPC: GetVersion

Server implements GetVersion returning a simple response; client calls it like a regular function.

Server Streaming RPC: DownloadLogs

Server yields LogEntry objects; client iterates over the response to read logs.

Client Streaming RPC: UploadFirmware

Server receives a request_iterator and returns a single UploadStatus. Client sends chunks via a generator.

def upload_firmware(stub, file_path, product_id):
    def chunk_generator():
        first_chunk = firmware_pb2.FirmwareChunk(product_id=product_id, data=b'')
        yield first_chunk
        with open(file_path, 'rb') as f:
            while True:
                data = f.read(64 * 1024)
                if not data:
                    break
                yield firmware_pb2.FirmwareChunk(data=data)

    try:
        response = stub.UploadFirmware(chunk_generator(), timeout=600)
        return response.code == 0
    except grpc.RpcError as e:
        print(f"Upload failed: {e.code()}")
        return False

Bidirectional Streaming RPC: MonitorUpgrade

Server iterates over incoming commands and yields progress reports simultaneously; client uses a generator for commands and iterates over responses.

Comparison of the Four Types

Feature	Unary RPC	Server Streaming	Client Streaming	Bidirectional Streaming
Request	1	1	Multiple (stream)	Multiple (stream)
Response	1	Multiple (stream)	1	Multiple (stream)
Client Impl.	Regular call	Iterate responses	Generator for requests	Generator + iterate
Server Impl.	Regular function	Generator for responses	Iterate requests	Iterate + generator
Typical Scenario	Query, config	Log download, data push	File upload, data colect	Real-time interaction, console

Firmware Upgrade Client Implementation

The following Python script demonstrates a complete firmware update client that leverages client streaming RPC for upload, followed by polling for device status using a unary RPC.

import os
import time
import hashlib
import logging
import grpc
import firmware_pb2
import firmware_pb2_grpc
from google.protobuf.timestamp_pb2 import Timestamp
from google.protobuf import empty_pb2

class FirmwareUpdater:
    CHUNK_SIZE = 1024 * 1024  # 1MB
    INITIAL_WAIT_SECONDS = 120
    MAX_POLL_ATTEMPTS = 120
    POLLING_INTERVAL_SECONDS = 5
    PROGRESS_LOG_INTERVAL = 12

    def __init__(self, server_addr, verbose=False):
        self.server_addr = server_addr
        self.verbose = verbose
        self.channel = None
        self.firmware_stub = None
        self.config_stub = None
        self.start_time = None
        self.connected = False
        self.logger = logging.getLogger(self.__class__.__name__)
        if self.verbose:
            self.logger.setLevel(logging.DEBUG)

    def connect(self):
        self.logger.info(f"Connecting to {self.server_addr}")
        max_retries = 3
        for retry in range(max_retries):
            try:
                self.channel = grpc.insecure_channel(
                    self.server_addr,
                    options=[
                        ('grpc.max_send_message_length', 100 * 1024 * 1024),
                        ('grpc.max_receive_message_length', 100 * 1024 * 1024),
                        ('grpc.keepalive_time_ms', 10000),
                        ('grpc.keepalive_timeout_ms', 5000),
                        ('grpc.keepalive_permit_without_calls', 1),
                    ]
                )
                self.firmware_stub = firmware_pb2_grpc.FirmwareServiceStub(self.channel)
                self.config_stub = firmware_pb2_grpc.TestInterfaceConfigurationServiceStub(self.channel)
                grpc.channel_ready_future(self.channel).result(timeout=10)
                self.connected = True
                self.logger.info(f"Connected to {self.server_addr}")
                if self._test_connection():
                    return True
            except Exception as e:
                self.logger.warning(f"Connection attempt {retry+1} failed: {e}")
                time.sleep(2)
        self.logger.error("Could not connect after multiple attempts")
        return False

    def _test_connection(self):
        try:
            response = self.firmware_stub.GetVersion(firmware_pb2.VersionRequest(), timeout=10)
            self.logger.info(f"Server version: {response.version}")
            return True
        except Exception as e:
            self.logger.warning(f"Connection test failed: {e}")
            return False

    def _compute_md5(self, filepath):
        md5 = hashlib.md5()
        with open(filepath, 'rb') as f:
            while chunk := f.read(8192):
                md5.update(chunk)
        return md5.hexdigest().lower()

    def upload_firmware(self, filepath, product_id, dut_position=0):
        if not os.path.exists(filepath):
            self.logger.error(f"File not found: {filepath}")
            return False
        filesize = os.path.getsize(filepath)
        filename = os.path.basename(filepath)
        filehash = self._compute_md5(filepath)
        self.logger.info(f"Uploading {filename} ({filesize} bytes)")

        def generate_requests():
            # First message with metadata
            info = firmware_pb2.FirmwareChunk(
                product_id=product_id,
                data=b'',
            )
            yield info

            with open(filepath, 'rb') as f:
                sent = 0
                while True:
                    chunk = f.read(self.CHUNK_SIZE)
                    if not chunk:
                        break
                    sent += len(chunk)
                    progress = (sent * 100) // filesize if filesize > 0 else 0
                    if sent % (10 * self.CHUNK_SIZE) == 0 or sent == filesize:
                        self.logger.info(f"Progress: {progress}%")
                    yield firmware_pb2.FirmwareChunk(data=chunk)

        try:
            response = self.firmware_stub.UploadFirmware(generate_requests(), timeout=600)
            if response.code == 0:
                self.logger.info(f"Upload successful: {response.message}")
                return True
            self.logger.error(f"Upload rejected: {response.message}")
            return False
        except grpc.RpcError as e:
            self.logger.error(f"Upload gRPC error: {e.code()} - {e.details()}")
            return False

    def wait_for_device(self, max_retries=None):
        if max_retries is None:
            max_retries = self.MAX_POLL_ATTEMPTS

        self.logger.info(f"Waiting {self.INITIAL_WAIT_SECONDS}s for device to begin upgrade...")
        time.sleep(self.INITIAL_WAIT_SECONDS)
        self.logger.info(f"Polling every {self.POLLING_INTERVAL_SECONDS}s (max {max_retries} attempts)...")

        for attempt in range(1, max_retries + 1):
            try:
                # Create a fresh channel for each poll attempt
                with grpc.insecure_channel(self.server_addr) as channel:
                    grpc.channel_ready_future(channel).result(timeout=self.POLLING_INTERVAL_SECONDS - 1)
                    temp_stub = firmware_pb2_grpc.TestInterfaceConfigurationServiceStub(channel)
                    response = temp_stub.GetSWInfos(empty_pb2.Empty(), timeout=2)

                if response is not None:
                    self.logger.info("Device is back online!")
                    for info in response.sw_info:
                        self.logger.info(f"  {info.sw_name}: {info.product_number}")
                    return True
            except grpc.RpcError as e:
                if e.code() not in (grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED):
                    self.logger.debug(f"Polling RPC error (attempt {attempt}): {e.details()}")
            except Exception as e:
                self.logger.debug(f"Polling exception (attempt {attempt}): {e}")

            if attempt % self.PROGRESS_LOG_INTERVAL == 0:
                self.logger.info(f"Still waiting... (attempt {attempt}/{max_retries})")
            time.sleep(self.POLLING_INTERVAL_SECONDS)

        self.logger.error("Device did not come back online within the polling window")
        return False

    def run_upgrade(self, filepath, product_id, dut_position=0):
        self.start_time = time.time()
        if not self.upload_firmware(filepath, product_id, dut_position):
            return False
        upload_time = time.time() - self.start_time
        self.logger.info(f"Upload took {upload_time:.2f}s")

        if not self.wait_for_device():
            return False

        total_time = time.time() - self.start_time
        self.logger.info(f"Firmware upgrade completed in {total_time:.2f}s")
        return True

    def close(self):
        if self.channel:
            self.channel.close()
            self.logger.info("Connection closed")
            self.connected = False

Key implementation details:

• Generator as request stream: The generate_requests function yields chunks one by one; the gRPC runtime sends them as HTTP/2 DATA frames.
• Stream termination: When the generator finishes (file fully read), gRPC automatically sends a half-close signal, informing the server that no more data is coming.
• Timeout control: The timeout parameter sets a deadline for the entire RPC, including all chunk transmissions and server processing.
• Error handling: Catching grpc.RpcError allows handling network issues, timeouts (DEADLINE_EXCEEDED), or server-returned errors.
• Polling with fresh channels: After upload, the device reboots; the client creates a new channel for each polling attempt to avoid issues with stale connections.

Python vs. C++ gRPC Summary

Feature	Python gRPC	C++ gRPC
Request Stream Implementation	Generator (yield). Implicit flow control.	ClientWriter object. Explicit Write(), WritesDone(), Finish().
Code Style	Concise, declarative, Pythonic.	Fine-grained, explicit, RAII.
Error Handling	Primarily exceptions (grpc.RpcError).	Check return value of Write(), examine Status from Finish().
Timeout Setting	timeout parameter in stub method calls.	ClientContext::set_deadline().
Resource Management	with statement and reference counting.	Smart pointers for ClientWriter, RAII.

Despite the stylistic differences, both Python and C++ implementations are built on the same gRPC C Core, ensuring identical network transport, flow control, and message serialization. Python’s generator model makes streaming programming intuitive and is well-suited for rapid development and client tools like the firmware upgrader.