Skip to content

pamqp.decode

pamqp.decode

Functions for decoding data of various types including field tables and arrays

by_type(value, data_type, offset=0)

Decodes values using the specified type

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required
data_type str

The data type name of the value

 required
offset int

The starting position of the data in the byte stream

 0

Raises:

Type Description
ValueError

when the data type is unknown
Source code in pamqp/decode.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def by_type(
    value: bytes, data_type: str, offset: int = 0
) -> tuple[int, common.FieldValue]:
    """Decodes values using the specified type

    :param value: The binary value to decode
    :param data_type: The data type name of the value
    :param offset: The starting position of the data in the byte stream
    :rtype: :class:`tuple` (:class:`int`, :const:`pamqp.common.FieldValue`)
    :raises ValueError: when the data type is unknown

    """
    if data_type == 'bit':
        return bit(value, offset)
    decoder = METHODS.get(data_type)
    if decoder is None:
        raise ValueError(f'Unknown type: {data_type}')
    return decoder(value)

bit(value, position)

Decode a bit value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required
position int

The position in the byte of the bit value

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def bit(value: bytes, position: int) -> tuple[int, bool]:
    """Decode a bit value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :param position: The position in the byte of the bit value
    :rtype: :class:`tuple` (:class:`int`, :class:`bool`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        bit_buffer = common.Struct.byte.unpack_from(value)[0]
        return 0, (bit_buffer & (1 << position)) != 0
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack bit value') from err

boolean(value)

Decode a boolean value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
50
51
52
53
54
55
56
57
58
59
60
61
def boolean(value: bytes) -> tuple[int, bool]:
    """Decode a boolean value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`bool`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 1, bool(common.Struct.byte.unpack_from(value[0:1])[0])
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack boolean value') from err

byte_array(value)

Decode a byte_array value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def byte_array(value: bytes) -> tuple[int, bytearray]:
    """Decode a byte_array value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`bytearray`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        length = common.Struct.integer.unpack(value[0:4])[0]
        return length + 4, bytearray(value[4 : length + 4])
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack byte array value') from err

decimal(value)

Decode a decimal value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def decimal(value: bytes) -> tuple[int, _decimal.Decimal]:
    """Decode a decimal value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`decimal.Decimal`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        decimals = common.Struct.byte.unpack(value[0:1])[0]
        raw = common.Struct.uint.unpack(value[1:5])[0]
        return 5, _decimal.Decimal(raw) * (_decimal.Decimal(10) ** -decimals)
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack decimal value') from err

double(value)

Decode a double value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def double(value: bytes) -> tuple[int, float]:
    """Decode a double value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`float`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 8, common.Struct.double.unpack_from(value)[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack double value') from err

floating_point(value)

Decode a floating point value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
109
110
111
112
113
114
115
116
117
118
119
120
def floating_point(value: bytes) -> tuple[int, float]:
    """Decode a floating point value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`float`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 4, common.Struct.float.unpack_from(value)[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack floating point value') from err

long_int(value)

Decode a long integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
123
124
125
126
127
128
129
130
131
132
133
134
def long_int(value: bytes) -> tuple[int, int]:
    """Decode a long integer value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 4, common.Struct.long.unpack(value[0:4])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack long integer value') from err

long_uint(value)

Decode an unsigned long integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def long_uint(value: bytes) -> tuple[int, int]:
    """Decode an unsigned long integer value, returning bytes consumed and
    the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 4, common.Struct.ulong.unpack(value[0:4])[0]
    except (struct.error, TypeError) as err:
        raise ValueError(
            'Could not unpack unsigned long integer value'
        ) from err

long_long_int(value)

Decode a long-long integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
def long_long_int(value: bytes) -> tuple[int, int]:
    """Decode a long-long integer value, returning bytes consumed and the
    value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 8, common.Struct.long_long_int.unpack(value[0:8])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack long-long integer value') from err

long_str(value)

Decode a string value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def long_str(value: bytes) -> tuple[int, str | bytes]:
    """Decode a string value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`str`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        length = common.Struct.integer.unpack(value[0:4])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack long string value') from err
    try:
        return length + 4, value[4 : length + 4].decode('utf-8')
    except UnicodeDecodeError:
        return length + 4, value[4 : length + 4]

octet(value)

Decode an octet value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
187
188
189
190
191
192
193
194
195
196
197
198
def octet(value: bytes) -> tuple[int, int]:
    """Decode an octet value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 1, common.Struct.byte.unpack(value[0:1])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack octet value') from err

short_int(value)

Decode a short integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
201
202
203
204
205
206
207
208
209
210
211
212
def short_int(value: bytes) -> tuple[int, int]:
    """Decode a short integer value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 2, common.Struct.short.unpack_from(value[0:2])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack short integer value') from err

short_uint(value)

Decode an unsigned short integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def short_uint(value: bytes) -> tuple[int, int]:
    """Decode an unsigned short integer value, returning bytes consumed and
    the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 2, common.Struct.ushort.unpack_from(value[0:2])[0]
    except (struct.error, TypeError) as err:
        raise ValueError(
            'Could not unpack unsigned short integer value'
        ) from err

short_short_int(value)

Decode a short-short integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
def short_short_int(value: bytes) -> tuple[int, int]:
    """Decode a short-short integer value, returning bytes consumed and the
    value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 1, common.Struct.short_short_int.unpack_from(value[0:1])[0]
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack short-short integer value') from err

short_short_uint(value)

Decode a unsigned short-short integer value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def short_short_uint(value: bytes) -> tuple[int, int]:
    """Decode a unsigned short-short integer value, returning bytes consumed
    and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`int`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        return 1, common.Struct.short_short_uint.unpack_from(value[0:1])[0]
    except (struct.error, TypeError) as err:
        raise ValueError(
            'Could not unpack unsigned short-short integer value'
        ) from err

short_str(value)

Decode a string value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
def short_str(value: bytes) -> tuple[int, str]:
    """Decode a string value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`str`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        length = common.Struct.byte.unpack(value[0:1])[0]
        return length + 1, value[1 : length + 1].decode('utf-8')
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack short string value') from err

timestamp(value)

Decode a timestamp value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def timestamp(value: bytes) -> tuple[int, datetime.datetime]:
    """Decode a timestamp value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :class:`datetime.datetime`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        temp = common.Struct.timestamp.unpack(value[0:8])
        ts_value = temp[0]

        # Anything above the year 2106 is likely milliseconds
        if ts_value > 0xFFFFFFFF:
            ts_value /= 1000.0

        return 8, datetime.datetime.fromtimestamp(ts_value, tz=datetime.UTC)
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack timestamp value') from err

embedded_value(value)

Dynamically decode a value based upon the starting byte

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def embedded_value(value: bytes) -> tuple[int, common.FieldValue]:
    """Dynamically decode a value based upon the starting byte

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :const:`pamqp.common.FieldValue`)
    :raises ValueError: when the binary data can not be unpacked

    """
    if not value:
        return 0, None
    try:
        bytes_consumed, temp = TABLE_MAPPING[value[0:1]](value[1:])
    except KeyError as err:
        raise ValueError(f'Unknown type: {value[:1]!r}') from err
    return bytes_consumed + 1, temp

field_array(value)

Decode a field array value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def field_array(value: bytes) -> tuple[int, common.FieldArray]:
    """Decode a field array value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :const:`pamqp.common.FieldArray`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        length = common.Struct.integer.unpack(value[0:4])[0]
        offset = 4
        data = []
        field_array_end = offset + length
        if field_array_end > len(value):
            raise ValueError('Field array length exceeds available data')
        while offset < field_array_end:
            consumed, result = embedded_value(value[offset:])
            offset += consumed
            data.append(result)
        return offset, data
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack data') from err

field_table(value)

Decode a field array value, returning bytes consumed and the value.

Parameters:

Name Type Description Default
value bytes

The binary value to decode

 required

Raises:

Type Description
ValueError

when the binary data can not be unpacked
Source code in pamqp/decode.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def field_table(value: bytes) -> tuple[int, common.FieldTable]:
    """Decode a field array value, returning bytes consumed and the value.

    :param value: The binary value to decode
    :rtype: :class:`tuple` (:class:`int`, :const:`pamqp.common.FieldTable`)
    :raises ValueError: when the binary data can not be unpacked

    """
    try:
        length = common.Struct.integer.unpack(value[0:4])[0]
        offset = 4
        data = {}
        field_table_end = offset + length
        if field_table_end > len(value):
            raise ValueError('Field table length exceeds available data')
        while offset < field_table_end:
            key_length = common.Struct.byte.unpack_from(value, offset)[0]
            offset += 1
            if offset + key_length > field_table_end:
                raise ValueError('Field table key length exceeds data')
            key = value[offset : offset + key_length].decode('utf-8')
            offset += key_length
            consumed, result = embedded_value(value[offset:])
            offset += consumed
            data[key] = result
        return field_table_end, data
    except (struct.error, TypeError) as err:
        raise ValueError('Could not unpack data') from err

void(_)

Return a void, no data to decode

Parameters:

Name Type Description Default
_ bytes

The empty bytes object to ignore

 required
Source code in pamqp/decode.py 
371
372
373
374
375
376
377
378
def void(_: bytes) -> tuple[int, None]:
    """Return a void, no data to decode

    :param _: The empty bytes object to ignore
    :rtype: :class:`tuple` (:class:`int`, :const:`None`)

    """
    return 0, None