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`

    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30

    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`

    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46

    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

        """
        bit_buffer = common.Struct.byte.unpack_from(value)[0]
        try:
            return 0, (bit_buffer & (1 << position)) != 0
        except 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`

    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60

    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 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`

    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75

    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 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`

    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91

    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.integer.unpack(value[1:5])[0]
            return 5, _decimal.Decimal(raw) * (_decimal.Decimal(10) ** -decimals)
        except 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`

     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105

    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 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`

    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119

    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 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`

    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133

    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 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`

    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150

    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 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`

    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165

    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 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`

    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183

    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 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`

    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197

    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 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`

    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211

    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 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`

    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228

    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 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`

    231
    232
    233
    234
    235
    236
    237
    238
    239
    240
    241
    242
    243

    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 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`

    246
    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257
    258
    259
    260

    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 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`

    263
    264
    265
    266
    267
    268
    269
    270
    271
    272
    273
    274
    275

    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 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`

    278
    279
    280
    281
    282
    283
    284
    285
    286
    287
    288
    289
    290
    291
    292
    293
    294
    295
    296

    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 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`

    299
    300
    301
    302
    303
    304
    305
    306
    307
    308
    309
    310
    311
    312
    313

    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`

    316
    317
    318
    319
    320
    321
    322
    323
    324
    325
    326
    327
    328
    329
    330
    331
    332
    333
    334
    335

    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
            while offset < field_array_end:
                consumed, result = embedded_value(value[offset:])
                offset += consumed
                data.append(result)
            return offset, data
        except 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`

    338
    339
    340
    341
    342
    343
    344
    345
    346
    347
    348
    349
    350
    351
    352
    353
    354
    355
    356
    357
    358
    359
    360
    361

    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
            while offset < field_table_end:
                key_length = common.Struct.byte.unpack_from(value, offset)[0]
                offset += 1
                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 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`

    364
    365
    366
    367
    368
    369
    370
    371

    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