Mock Protobuf Binary Encoding in Python
Demonstrates a minimal protobuf-like binary encoding and decoding of an event dataclass using varints and length-delimited fields in pure Python.
Python code
79 linesimport struct
from dataclasses import dataclass
@dataclass
class Event:
id: int
user_id: int
action: str
def encode(self) -> bytes:
# Mock protobuf-like binary encoding using varint and length-delimited fields
buf = bytearray()
# field 1: varint id (tag = (1 << 3) | 0 = 8)
buf.append(8)
buf.extend(_encode_varint(self.id))
# field 2: varint user_id (tag = (2 << 3) | 0 = 16)
buf.append(16)
buf.extend(_encode_varint(self.user_id))
# field 3: string action (tag = (3 << 3) | 2 = 26)
buf.append(26)
action_bytes = self.action.encode("utf-8")
buf.extend(_encode_varint(len(action_bytes)))
buf.extend(action_bytes)
return bytes(buf)
def _encode_varint(value: int) -> bytes:
if value < 0:
raise ValueError("only non-negative integers supported")
result = bytearray()
while value > 127:
result.append((value & 0x7F) | 0x80)
value >>= 7
result.append(value)
return bytes(result)
def decode(binary: bytes) -> dict:
# Minimal decoder for the mock format (supports fields 1, 2, 3)
offset = 0
result = {}
while offset < len(binary):
tag = binary[offset]
offset += 1
field_number = tag >> 3
wire_type = tag & 0x07
if wire_type == 0:
value, offset = _decode_varint(binary, offset)
result[field_number] = value
elif wire_type == 2:
length, offset = _decode_varint(binary, offset)
value = binary[offset : offset + length].decode("utf-8")
offset += length
result[field_number] = value
else:
raise ValueError(f"unsupported wire type {wire_type}")
return result
def _decode_varint(data: bytes, offset: int) -> tuple[int, int]:
result = 0
shift = 0
while True:
byte = data[offset]
offset += 1
result |= (byte & 0x7F) << shift
if not byte & 0x80:
break
shift += 7
return result, offset
if __name__ == "__main__":
event = Event(id=150, user_id=42, action="login")
encoded = event.encode()
print(f"Encoded bytes: {encoded.hex()}")
decoded = decode(encoded)
print(f"Decoded dict: {decoded}")
Output
Encoded bytes: 089601102a1a056c6f67696e
Decoded dict: {1: 150, 2: 42, 3: 'login'}
How it works
The encode method builds a bytearray that mirrors protobuf's wire format. Field tags combine the field number and wire type (varint=0, length-delimited=2) by shifting the field number left 3 bits. _encode_varint handles multi-byte varints by setting the continuation bit (0x80) on every byte except the last. The decoder walks the same format, reading tags, matching wire types, and reconstructing values; it stays minimal and only supports fields 1–3. This approach shows the essence of protobuf encoding without needing the protobuf package.
Common mistakes
- Forgetting to set the continuation bit (0x80) on non-final varint bytes
- Confusing the field tag with the field number (tag combines number and wire type)
- Missing UTF-8 decode step for length-delimited string fields
- Assuming negative integers can be encoded as varints (must raise or use zigzag)
Variations
- Use the official `google.protobuf` library to generate and serialize real `.proto` messages.
- Implement a general-purpose varint encoder/decoder that supports zigzag encoding for negative numbers.
Real-world use cases
- Understanding the wire format when debugging why two services disagree on protobuf field ordering.
- Building a lightweight binary protocol for embedded devices where the full protobuf runtime is too heavy.
- Teaching or onboarding new engineers on how Kafka or gRPC payloads are actually serialized under the hood.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.