0
# Types
1
2
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.
3
4
## Capabilities
5
6
### Extension Type
7
8
Container for custom binary data with type codes, enabling serialization of application-specific data structures within MessagePack format.
9
10
```python { .api }
11
class ExtType:
12
def __init__(self, code, data):
13
"""
14
ExtType represents ext type in msgpack.
15
16
Parameters:
17
- code: int, extension type code (0-127)
18
- data: bytes, extension data
19
20
Raises:
21
TypeError: When code is not int or data is not bytes
22
ValueError: When code is not in range 0-127
23
"""
24
25
code: int
26
"""Extension type code (0-127)"""
27
28
data: bytes
29
"""Extension binary data"""
30
```
31
32
ExtType is implemented as a named tuple with validation in the constructor.
33
34
### Timestamp Type
35
36
High-precision timestamp representation supporting nanosecond accuracy and efficient binary encoding for time-sensitive applications.
37
38
```python { .api }
39
class Timestamp:
40
def __init__(self, seconds, nanoseconds=0):
41
"""
42
Initialize a Timestamp object.
43
44
Parameters:
45
- seconds: int, seconds since UNIX epoch (may be negative)
46
- nanoseconds: int, nanoseconds to add (0-999999999, default: 0)
47
48
Raises:
49
TypeError: When parameters are not integers
50
ValueError: When nanoseconds out of range
51
"""
52
53
seconds: int
54
"""Number of seconds since UNIX epoch"""
55
56
nanoseconds: int
57
"""Number of nanoseconds (0-999999999)"""
58
59
@staticmethod
60
def from_bytes(b):
61
"""
62
Unpack bytes into a Timestamp object.
63
64
Parameters:
65
- b: bytes, payload from msgpack ext message with code -1
66
67
Returns:
68
Timestamp: Unpacked timestamp object
69
70
Raises:
71
ValueError: When byte length is not 4, 8, or 12
72
"""
73
74
def to_bytes(self):
75
"""
76
Pack this Timestamp object into bytes.
77
78
Returns:
79
bytes: Payload for EXT message with code -1
80
"""
81
82
@staticmethod
83
def from_unix(unix_sec):
84
"""
85
Create a Timestamp from posix timestamp in seconds.
86
87
Parameters:
88
- unix_sec: int or float, posix timestamp in seconds
89
90
Returns:
91
Timestamp: New timestamp object
92
"""
93
94
def to_unix(self):
95
"""
96
Get the timestamp as a floating-point value.
97
98
Returns:
99
float: Posix timestamp in seconds
100
"""
101
102
@staticmethod
103
def from_unix_nano(unix_ns):
104
"""
105
Create a Timestamp from posix timestamp in nanoseconds.
106
107
Parameters:
108
- unix_ns: int, posix timestamp in nanoseconds
109
110
Returns:
111
Timestamp: New timestamp object
112
"""
113
114
def to_unix_nano(self):
115
"""
116
Get the timestamp as a unixtime in nanoseconds.
117
118
Returns:
119
int: Posix timestamp in nanoseconds
120
"""
121
122
@staticmethod
123
def from_datetime(dt):
124
"""
125
Create a Timestamp from datetime with tzinfo.
126
127
Parameters:
128
- dt: datetime.datetime, datetime object with timezone
129
130
Returns:
131
Timestamp: New timestamp object
132
"""
133
134
def to_datetime(self):
135
"""
136
Get the timestamp as a UTC datetime.
137
138
Returns:
139
datetime.datetime: UTC datetime object
140
"""
141
142
def __repr__(self):
143
"""String representation of Timestamp."""
144
145
def __eq__(self, other):
146
"""Check for equality with another Timestamp object."""
147
148
def __ne__(self, other):
149
"""Check for inequality with another Timestamp object."""
150
151
def __hash__(self):
152
"""Hash value for use in sets and dicts."""
153
```
154
155
## Usage Examples
156
157
### Custom Extension Types
158
159
```python
160
import msgpack
161
import array
162
163
def pack_array(obj):
164
"""Custom packer for Python arrays."""
165
if isinstance(obj, array.array) and obj.typecode == 'd':
166
return msgpack.ExtType(42, obj.tobytes())
167
raise TypeError(f"Unknown type: {type(obj)}")
168
169
def unpack_array(code, data):
170
"""Custom unpacker for Python arrays."""
171
if code == 42:
172
arr = array.array('d')
173
arr.frombytes(data)
174
return arr
175
return msgpack.ExtType(code, data)
176
177
# Pack array as extension type
178
data = array.array('d', [1.2, 3.4, 5.6])
179
packed = msgpack.packb(data, default=pack_array)
180
181
# Unpack extension type back to array
182
unpacked = msgpack.unpackb(packed, ext_hook=unpack_array)
183
print(unpacked) # array('d', [1.2, 3.4, 5.6])
184
```
185
186
### Working with Timestamps
187
188
```python
189
import msgpack
190
import datetime
191
192
# Create timestamp from current time
193
now = datetime.datetime.now(datetime.timezone.utc)
194
ts = msgpack.Timestamp.from_datetime(now)
195
196
# Pack timestamp
197
packed = msgpack.packb(ts)
198
199
# Unpack timestamp
200
unpacked = msgpack.unpackb(packed)
201
print(f"Seconds: {unpacked.seconds}, Nanoseconds: {unpacked.nanoseconds}")
202
203
# Convert back to datetime
204
restored_dt = unpacked.to_datetime()
205
print(f"Original: {now}")
206
print(f"Restored: {restored_dt}")
207
```
208
209
### High-Precision Timestamps
210
211
```python
212
import msgpack
213
import time
214
215
# Create timestamp with nanosecond precision
216
nano_time = int(time.time_ns())
217
ts = msgpack.Timestamp.from_unix_nano(nano_time)
218
219
# Verify round-trip precision
220
packed = msgpack.packb(ts)
221
unpacked = msgpack.unpackb(packed)
222
restored_nano = unpacked.to_unix_nano()
223
224
print(f"Original: {nano_time}")
225
print(f"Restored: {restored_nano}")
226
print(f"Exact match: {nano_time == restored_nano}")
227
```
228
229
### Binary Format Details
230
231
```python
232
import msgpack
233
234
# Timestamp encoding formats
235
ts1 = msgpack.Timestamp(1234567890, 0) # 32-bit format
236
ts2 = msgpack.Timestamp(1234567890, 123456789) # 64-bit format
237
ts3 = msgpack.Timestamp(-1, 123456789) # 96-bit format
238
239
# Check binary sizes
240
print(f"32-bit timestamp: {len(ts1.to_bytes())} bytes") # 4 bytes
241
print(f"64-bit timestamp: {len(ts2.to_bytes())} bytes") # 8 bytes
242
print(f"96-bit timestamp: {len(ts3.to_bytes())} bytes") # 12 bytes
243
244
# ExtType with custom data
245
ext = msgpack.ExtType(100, b'custom binary data')
246
packed = msgpack.packb(ext)
247
unpacked = msgpack.unpackb(packed)
248
print(f"ExtType code: {unpacked.code}, data: {unpacked.data}")
249
```