tessl/pypi-msgpack
MessagePack serializer for efficient binary data exchange among multiple languages
—
Pending
types.mddocs/
Types
Specialized types for handling extension data and timestamps in MessagePack. These classes enable custom serialization of complex data structures and provide built-in support for high-precision timestamps.
Capabilities
Extension Type
Container for custom binary data with type codes, enabling serialization of application-specific data structures within MessagePack format.
class ExtType:
def __init__(self, code, data):
"""
ExtType represents ext type in msgpack.
Parameters:
- code: int, extension type code (0-127)
- data: bytes, extension data
Raises:
TypeError: When code is not int or data is not bytes
ValueError: When code is not in range 0-127
"""
code: int
"""Extension type code (0-127)"""
data: bytes
"""Extension binary data"""ExtType is implemented as a named tuple with validation in the constructor.
Timestamp Type
High-precision timestamp representation supporting nanosecond accuracy and efficient binary encoding for time-sensitive applications.
class Timestamp:
def __init__(self, seconds, nanoseconds=0):
"""
Initialize a Timestamp object.
Parameters:
- seconds: int, seconds since UNIX epoch (may be negative)
- nanoseconds: int, nanoseconds to add (0-999999999, default: 0)
Raises:
TypeError: When parameters are not integers
ValueError: When nanoseconds out of range
"""
seconds: int
"""Number of seconds since UNIX epoch"""
nanoseconds: int
"""Number of nanoseconds (0-999999999)"""
@staticmethod
def from_bytes(b):
"""
Unpack bytes into a Timestamp object.
Parameters:
- b: bytes, payload from msgpack ext message with code -1
Returns:
Timestamp: Unpacked timestamp object
Raises:
ValueError: When byte length is not 4, 8, or 12
"""
def to_bytes(self):
"""
Pack this Timestamp object into bytes.
Returns:
bytes: Payload for EXT message with code -1
"""
@staticmethod
def from_unix(unix_sec):
"""
Create a Timestamp from posix timestamp in seconds.
Parameters:
- unix_sec: int or float, posix timestamp in seconds
Returns:
Timestamp: New timestamp object
"""
def to_unix(self):
"""
Get the timestamp as a floating-point value.
Returns:
float: Posix timestamp in seconds
"""
@staticmethod
def from_unix_nano(unix_ns):
"""
Create a Timestamp from posix timestamp in nanoseconds.
Parameters:
- unix_ns: int, posix timestamp in nanoseconds
Returns:
Timestamp: New timestamp object
"""
def to_unix_nano(self):
"""
Get the timestamp as a unixtime in nanoseconds.
Returns:
int: Posix timestamp in nanoseconds
"""
@staticmethod
def from_datetime(dt):
"""
Create a Timestamp from datetime with tzinfo.
Parameters:
- dt: datetime.datetime, datetime object with timezone
Returns:
Timestamp: New timestamp object
"""
def to_datetime(self):
"""
Get the timestamp as a UTC datetime.
Returns:
datetime.datetime: UTC datetime object
"""
def __repr__(self):
"""String representation of Timestamp."""
def __eq__(self, other):
"""Check for equality with another Timestamp object."""
def __ne__(self, other):
"""Check for inequality with another Timestamp object."""
def __hash__(self):
"""Hash value for use in sets and dicts."""Usage Examples
Custom Extension Types
import msgpack
import array
def pack_array(obj):
"""Custom packer for Python arrays."""
if isinstance(obj, array.array) and obj.typecode == 'd':
return msgpack.ExtType(42, obj.tobytes())
raise TypeError(f"Unknown type: {type(obj)}")
def unpack_array(code, data):
"""Custom unpacker for Python arrays."""
if code == 42:
arr = array.array('d')
arr.frombytes(data)
return arr
return msgpack.ExtType(code, data)
# Pack array as extension type
data = array.array('d', [1.2, 3.4, 5.6])
packed = msgpack.packb(data, default=pack_array)
# Unpack extension type back to array
unpacked = msgpack.unpackb(packed, ext_hook=unpack_array)
print(unpacked) # array('d', [1.2, 3.4, 5.6])Working with Timestamps
import msgpack
import datetime
# Create timestamp from current time
now = datetime.datetime.now(datetime.timezone.utc)
ts = msgpack.Timestamp.from_datetime(now)
# Pack timestamp
packed = msgpack.packb(ts)
# Unpack timestamp
unpacked = msgpack.unpackb(packed)
print(f"Seconds: {unpacked.seconds}, Nanoseconds: {unpacked.nanoseconds}")
# Convert back to datetime
restored_dt = unpacked.to_datetime()
print(f"Original: {now}")
print(f"Restored: {restored_dt}")High-Precision Timestamps
import msgpack
import time
# Create timestamp with nanosecond precision
nano_time = int(time.time_ns())
ts = msgpack.Timestamp.from_unix_nano(nano_time)
# Verify round-trip precision
packed = msgpack.packb(ts)
unpacked = msgpack.unpackb(packed)
restored_nano = unpacked.to_unix_nano()
print(f"Original: {nano_time}")
print(f"Restored: {restored_nano}")
print(f"Exact match: {nano_time == restored_nano}")Binary Format Details
import msgpack
# Timestamp encoding formats
ts1 = msgpack.Timestamp(1234567890, 0) # 32-bit format
ts2 = msgpack.Timestamp(1234567890, 123456789) # 64-bit format
ts3 = msgpack.Timestamp(-1, 123456789) # 96-bit format
# Check binary sizes
print(f"32-bit timestamp: {len(ts1.to_bytes())} bytes") # 4 bytes
print(f"64-bit timestamp: {len(ts2.to_bytes())} bytes") # 8 bytes
print(f"96-bit timestamp: {len(ts3.to_bytes())} bytes") # 12 bytes
# ExtType with custom data
ext = msgpack.ExtType(100, b'custom binary data')
packed = msgpack.packb(ext)
unpacked = msgpack.unpackb(packed)
print(f"ExtType code: {unpacked.code}, data: {unpacked.data}")Install with Tessl CLI
npx tessl i tessl/pypi-msgpack