MessagePack serializer for efficient binary data exchange among multiple languages
—
Comprehensive exception hierarchy for robust error handling during MessagePack serialization and deserialization operations. These exceptions provide detailed information about various failure modes and enable precise error recovery strategies.
Root exception types that serve as base classes for more specific MessagePack errors.
class UnpackException(Exception):
"""
Base class for some exceptions raised while unpacking.
Note: unpack may raise exception other than subclass of
UnpackException. If you want to catch all errors, catch
Exception instead.
"""Exceptions related to internal buffer state and capacity limits during streaming operations.
class BufferFull(UnpackException):
"""
Raised when internal buffer exceeds size limits.
This exception occurs when the unpacker's internal buffer
reaches the configured max_buffer_size limit.
"""
class OutOfData(UnpackException):
"""
Raised when buffer contains incomplete data.
This exception occurs during streaming unpacking when
the current buffer doesn't contain enough data to
complete the requested operation.
"""Exceptions that indicate problems with MessagePack binary format compliance.
class FormatError(ValueError, UnpackException):
"""
Invalid msgpack format.
Raised when the binary data doesn't conform to the
MessagePack specification or contains corrupted data.
"""
class StackError(ValueError, UnpackException):
"""
Too nested data structure.
Raised when unpacking data with nesting levels that
exceed the recursion limits, preventing stack overflow.
"""Exceptions that handle cases where unpacking succeeds but additional data is present.
class ExtraData(UnpackValueError):
"""
ExtraData is raised when there is trailing data.
This exception is raised during one-shot (not streaming)
unpack operations when the input contains more data
than expected after successfully unpacking an object.
"""
unpacked: object
"""The successfully unpacked object"""
extra: bytes
"""The extra bytes that were not consumed"""
def __init__(self, unpacked, extra):
"""
Initialize ExtraData exception.
Parameters:
- unpacked: object, the successfully unpacked data
- extra: bytes, the remaining unconsumed data
"""
def __str__(self):
"""Return error message string."""Backward compatibility aliases for deprecated exception names.
UnpackValueError = ValueError
"""Deprecated. Use ValueError instead."""
PackException = Exception
"""Deprecated. Use Exception instead to catch all packing exceptions."""
PackValueError = ValueError
"""Deprecated. Use ValueError instead."""
PackOverflowError = OverflowError
"""Deprecated. Use OverflowError instead."""import msgpack
try:
data = msgpack.unpackb(b'\x81\xa4name\xa5Alice')
print(data) # {'name': 'Alice'}
except msgpack.FormatError as e:
print(f"Invalid MessagePack format: {e}")
except msgpack.UnpackException as e:
print(f"Unpacking error: {e}")import msgpack
# Data with extra bytes at the end
data_with_extra = msgpack.packb('hello') + msgpack.packb('world')
try:
result = msgpack.unpackb(data_with_extra)
except msgpack.ExtraData as e:
print(f"Unpacked: {e.unpacked}") # 'hello'
print(f"Extra bytes: {len(e.extra)}") # Length of packed 'world'
# Process the extra data
remaining = msgpack.unpackb(e.extra)
print(f"Remaining: {remaining}") # 'world'import msgpack
unpacker = msgpack.Unpacker(max_buffer_size=1024)
while True:
try:
chunk = receive_data() # Get data from network/file
unpacker.feed(chunk)
while True:
try:
obj = unpacker.unpack()
process_object(obj)
except msgpack.OutOfData:
# Need more data, break inner loop
break
except msgpack.BufferFull:
print("Buffer full, clearing and restarting")
unpacker = msgpack.Unpacker(max_buffer_size=1024)
except msgpack.FormatError as e:
print(f"Format error: {e}")
# Skip corrupted data or reset connection
unpacker = msgpack.Unpacker(max_buffer_size=1024)
except ConnectionError:
print("Connection lost")
breakimport msgpack
def safe_unpack(data, max_size=1024*1024):
"""Safely unpack data with comprehensive error handling."""
try:
return msgpack.unpackb(
data,
max_buffer_size=max_size,
strict_map_key=True,
raw=False
)
except msgpack.BufferFull:
raise ValueError(f"Data exceeds size limit of {max_size} bytes")
except msgpack.FormatError:
raise ValueError("Invalid or corrupted MessagePack data")
except msgpack.StackError:
raise ValueError("Data structure too deeply nested")
except msgpack.ExtraData as e:
# Log warning but return the valid data
print(f"Warning: {len(e.extra)} extra bytes ignored")
return e.unpacked
# Usage
try:
obj = safe_unpack(untrusted_data)
process_trusted_object(obj)
except ValueError as e:
print(f"Unsafe data rejected: {e}")import msgpack
def robust_msgpack_operation(data):
"""Handle all possible MessagePack exceptions."""
try:
return msgpack.unpackb(data)
except msgpack.ExtraData as e:
# Handle extra data specifically
return e.unpacked
except msgpack.FormatError:
# Handle format errors
raise ValueError("Invalid MessagePack format")
except msgpack.StackError:
# Handle nesting errors
raise ValueError("Data too deeply nested")
except msgpack.BufferFull:
# Handle buffer overflow
raise ValueError("Data too large")
except msgpack.UnpackException:
# Catch any other unpacking errors
raise ValueError("Unknown unpacking error")
except Exception:
# Catch any other unexpected errors
raise ValueError("Unexpected error during unpacking")Install with Tessl CLI
npx tessl i tessl/pypi-msgpack