Parse and build binary data in Python
Company: OpenAI
Role: Software Engineer
Category: Data Manipulation (SQL/Python)
Difficulty: medium
Interview Round: Technical Screen
Using provided interfaces ByteReader(read(n), read_uint32_le, read_string) and ByteWriter(write(b), write_uint32_le, write_string), implement functions to pack and unpack messages for a simple binary protocol: message = {id:uint32 LE, payload_len:uint32 LE, payload:bytes}. Write parse_message(reader)->Message and build_message(writer, Message)->None with error handling for short reads, invalid lengths, and overflow. Avoid printing for debugging; design tests instead, and explain how you would verify correctness and performance.
Overview: This question evaluates proficiency in binary data handling and serialization/deserialization in Python, including little-endian integer encoding, length-prefixed payloads, robust error handling for short reads and invalid lengths, and testability and performance considerations.
Community answers
Answer by wingpad
from dataclasses import dataclass
@dataclass
class Message:
id: int
payload: bytes
--- Error Definitions ---
class ProtocolError(Exception):
"""Base exception for all protocol violations."""
pass
class ShortReadError(ProtocolError):
"""Raised when the stream ends before expected data is fully read."""
pass
class InvalidLengthError(ProtocolError):
"""Raised when a payload length is negative or exceeds allowed limits."""
pass
--- Core Logic ---
def parse_message(reader) -> Message:
"""
Parses a single Message from the reader.
Protocol: id (uint32 LE) + payload_len (uint32 LE) + payload (bytes)
"""
# 1. Read the Message ID
try:
msg_id = reader.read_uint32_le()
except (EOFError, IndexError) as e:
raise ShortReadError("Failed to read message ID from stream.") from e
# 2. Read the Payload Length
try:
payload_len = reader.read_uint32_le()
except (EOFError, IndexError) as e:
raise ShortReadError("Failed to read payload length from stream.") from e
# Sanity check: Enforce an upper bound limit (e.g., 10MB) to prevent OOM attacks
MAX_PAYLOAD_SIZE = 10 1024 1024
if payload_len > MAX_PAYLOAD_SIZE:
raise InvalidLengthError(f"Payload length {payload_len} exceeds maximum allowed limit.")
# 3. Read the Payload
try:
payload = reader.read(payload_len)
except (EOFError, IndexError) as e:
raise ShortReadError("Stream ended before full payload could be read.") from e
if len(payload) < payload_len:
raise ShortReadError(f"Expected {payload_len} bytes, but only retrieved {len(payload)} bytes.")
retur