Skip to content

# pamqp.encode

## `pamqp.encode`

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

### `DEPRECATED_RABBITMQ_SUPPORT = False` `module-attribute`

Toggle to support older versions of RabbitMQ.

### `support_deprecated_rabbitmq(enabled=True)`

Toggle the data types available in field-tables

If called with `True`, than RabbitMQ versions, the field-table integer types will not support the full AMQP spec.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`enabled` | `bool` | Specify if deprecated RabbitMQ versions are supported | `True`

Source code in `pamqp/encode.py`

    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33

    def support_deprecated_rabbitmq(enabled: bool = True) -> None:
        """Toggle the data types available in field-tables

        If called with `True`, than RabbitMQ versions, the field-table integer
        types will not support the full AMQP spec.

        :param enabled: Specify if deprecated RabbitMQ versions are supported

        """
        global DEPRECATED_RABBITMQ_SUPPORT

        DEPRECATED_RABBITMQ_SUPPORT = enabled

### `by_type(value, data_type)`

Takes a value of any type and tries to encode it with the specified encoder.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `FieldValue` | The value to encode | required
`data_type` | `str` | The data type name to use for encoding | required

Raises:

Type | Description
--- | ---
`TypeError` | when the :data:`data_type` is unknown

Source code in `pamqp/encode.py`

    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49

    def by_type(value: common.FieldValue, data_type: str) -> bytes:
        """Takes a value of any type and tries to encode it with the specified
        encoder.

        :param value: The value to encode
        :type value: :const:`pamqp.common.FieldValue`
        :param data_type: The data type name to use for encoding
        :raises TypeError: when the :data:`data_type` is unknown

        """
        try:
            return METHODS[str(data_type)](value)
        except KeyError as err:
            raise TypeError(f'Unknown type: {data_type}') from err

### `bit(value, byte, position)`

Encode a bit value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required
`byte` | `int` | The byte to apply the value to | required
`position` | `int` | The position in the byte to set the bit on | required

Source code in `pamqp/encode.py`

    52
    53
    54
    55
    56
    57
    58
    59
    60

    def bit(value: int, byte: int, position: int) -> int:
        """Encode a bit value

        :param value: Value to encode
        :param byte: The byte to apply the value to
        :param position: The position in the byte to set the bit on

        """
        return byte | (value << position)

### `boolean(value)`

Encode a boolean value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `bool` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    63
    64
    65
    66
    67
    68
    69
    70
    71
    72

    def boolean(value: bool) -> bytes:
        """Encode a boolean value

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, bool):
            raise TypeError(f'bool required, received {type(value)}')
        return common.Struct.short_short_uint.pack(int(value))

### `byte_array(value)`

Encode a byte array value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `bytearray` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    75
    76
    77
    78
    79
    80
    81
    82
    83
    84

    def byte_array(value: bytearray) -> bytes:
        """Encode a byte array value

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, bytearray):
            raise TypeError(f'bytearray required, received {type(value)}')
        return common.Struct.integer.pack(len(value)) + value

### `decimal(value)`

Encode a decimal.Decimal value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `Decimal` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

     87
     88
     89
     90
     91
     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102

    def decimal(value: _decimal.Decimal) -> bytes:
        """Encode a decimal.Decimal value

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, _decimal.Decimal):
            raise TypeError(f'decimal.Decimal required, received {type(value)}')
        tmp = str(value)
        if '.' in tmp:
            decimals = len(tmp.split('.')[-1])
            value = value.normalize()
            raw = int(value * (_decimal.Decimal(10) ** decimals))
            return struct.pack('>Bi', decimals, raw)
        return struct.pack('>Bi', 0, int(value))

### `double(value)`

Encode a floating point value as a double

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `float` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    105
    106
    107
    108
    109
    110
    111
    112
    113
    114

    def double(value: float) -> bytes:
        """Encode a floating point value as a double

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, float):
            raise TypeError(f'float required, received {type(value)}')
        return common.Struct.double.pack(value)

### `floating_point(value)`

Encode a floating point value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `float` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    117
    118
    119
    120
    121
    122
    123
    124
    125
    126

    def floating_point(value: float) -> bytes:
        """Encode a floating point value

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, float):
            raise TypeError(f'float required, received {type(value)}')
        return common.Struct.float.pack(value)

### `long_int(value)`

Encode a long integer

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141

    def long_int(value: int) -> bytes:
        """Encode a long integer

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        elif not (-2147483648 <= value <= 2147483647):
            raise TypeError('Long integer range: -2147483648 to 2147483647')
        return common.Struct.long.pack(value)

### `long_uint(value)`

Encode a long unsigned integer

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156

    def long_uint(value: int) -> bytes:
        """Encode a long unsigned integer

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        elif not (0 <= value <= 4294967295):
            raise TypeError('Long unsigned-integer range: 0 to 4294967295')
        return common.Struct.ulong.pack(value)

### `long_long_int(value)`

Encode a long-long int

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174

    def long_long_int(value: int) -> bytes:
        """Encode a long-long int

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        elif not (-9223372036854775808 <= value <= 9223372036854775807):
            raise TypeError(
                'long-long integer range: '
                '-9223372036854775808 to 9223372036854775807'
            )
        return common.Struct.long_long_int.pack(value)

### `long_string(value)`

Encode a "long string"

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `str` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    177
    178
    179
    180
    181
    182
    183
    184

    def long_string(value: str) -> bytes:
        """Encode a "long string"

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        return _string(common.Struct.integer, value)

### `octet(value)`

Encode an octet value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    187
    188
    189
    190
    191
    192
    193
    194
    195
    196

    def octet(value: int) -> bytes:
        """Encode an octet value

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        return common.Struct.byte.pack(value)

### `short_int(value)`

Encode a short integer

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

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

    def short_int(value: int) -> bytes:
        """Encode a short integer

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        elif not (-32768 <= value <= 32767):
            raise TypeError('Short integer range: -32678 to 32767')
        return common.Struct.short.pack(value)

### `short_uint(value)`

Encode an unsigned short integer

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

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

    def short_uint(value: int) -> bytes:
        """Encode an unsigned short integer

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if not isinstance(value, int):
            raise TypeError(f'int required, received {type(value)}')
        elif not (0 <= value <= 65535):
            raise TypeError('Short unsigned integer range: 0 to 65535')
        return common.Struct.ushort.pack(value)

### `short_string(value)`

Encode a string

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `str` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    229
    230
    231
    232
    233
    234
    235
    236

    def short_string(value: str) -> bytes:
        """Encode a string

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        return _string(common.Struct.byte, value)

### `timestamp(value)`

Encode a datetime.datetime object or time.struct_time

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `datetime | struct_time` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    239
    240
    241
    242
    243
    244
    245
    246
    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257

    def timestamp(
        value: datetime.datetime | time.struct_time,
    ) -> bytes:
        """Encode a datetime.datetime object or time.struct_time

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type

        """
        if isinstance(value, datetime.datetime):
            if value.tzinfo is None or value.tzinfo.utcoffset(value) is None:
                # assume datetime object is UTC
                value = value.replace(tzinfo=datetime.UTC)
            return common.Struct.timestamp.pack(int(value.timestamp()))
        if isinstance(value, time.struct_time):
            return common.Struct.timestamp.pack(calendar.timegm(value))
        raise TypeError(
            f'datetime.datetime or time.struct_time required, received {type(value)}'
        )

### `field_array(value)`

Encode a field array from a list of values

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `FieldArray` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    260
    261
    262
    263
    264
    265
    266
    267
    268
    269
    270
    271
    272
    273
    274

    def field_array(value: common.FieldArray) -> bytes:
        """Encode a field array from a list of values

        :param value: Value to encode
        :type value: :const:`pamqp.common.FieldArray`
        :raises TypeError: when the value is not the correct type

        """
        if not isinstance(value, list):
            raise TypeError(f'list of values required, received {type(value)}')
        data = []
        for item in value:
            data.append(encode_table_value(item))
        output = b''.join(data)
        return common.Struct.integer.pack(len(output)) + output

### `field_table(value)`

Encode a field table from a dict

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `FieldTable` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type

Source code in `pamqp/encode.py`

    277
    278
    279
    280
    281
    282
    283
    284
    285
    286
    287
    288
    289
    290
    291
    292
    293
    294
    295
    296
    297
    298
    299
    300

    def field_table(value: common.FieldTable) -> bytes:
        """Encode a field table from a dict

        :param value: Value to encode
        :type value: :const:`pamqp.common.FieldTable`
        :raises TypeError: when the value is not the correct type

        """
        if not value:  # If there is no value, return a standard 4 null bytes
            return common.Struct.integer.pack(0)
        elif not isinstance(value, dict):
            raise TypeError(f'dict required, received {type(value)}')
        data = []
        for key, item in sorted(value.items()):
            if len(key) > 128:  # field names have 128 char max
                LOGGER.warning('Truncating key %s to 128 bytes', key)
                key = key[0:128]
            data.append(short_string(key))
            try:
                data.append(encode_table_value(item))
            except TypeError as err:
                raise TypeError(f'{key} error: {err}') from err
        output = b''.join(data)
        return common.Struct.integer.pack(len(output)) + output

### `table_integer(value)`

Determines the best type of numeric type to encode value as, preferring the smallest data size first.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `int` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the value is not the correct type or outside the acceptable range for the data type

Source code in `pamqp/encode.py`

    303
    304
    305
    306
    307
    308
    309
    310
    311
    312
    313
    314
    315
    316
    317
    318
    319
    320
    321
    322
    323
    324
    325
    326

    def table_integer(value: int) -> bytes:
        """Determines the best type of numeric type to encode value as, preferring
        the smallest data size first.

        :param value: Value to encode
        :raises TypeError: when the value is not the correct type or outside the
            acceptable range for the data type

        """
        if DEPRECATED_RABBITMQ_SUPPORT:
            return _deprecated_table_integer(value)
        if -128 <= value <= 127:
            return b'b' + common.Struct.short_short_int.pack(value)
        elif -32768 <= value <= 32767:
            return b's' + short_int(value)
        elif 0 <= value <= 65535:
            return b'u' + short_uint(value)
        elif -2147483648 <= value <= 2147483647:
            return b'I' + long_int(value)
        elif 0 <= value <= 4294967295:
            return b'i' + long_uint(value)
        elif -9223372036854775808 <= value <= 9223372036854775807:
            return b'l' + long_long_int(value)
        raise TypeError(f'Unsupported numeric value: {value}')

### `encode_table_value(value)`

Takes a value of any type and tries to encode it with the proper encoder

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`value` | `FieldArray | FieldTable | FieldValue` | Value to encode | required

Raises:

Type | Description
--- | ---
`TypeError` | when the type of the value is not supported

Source code in `pamqp/encode.py`

    361
    362
    363
    364
    365
    366
    367
    368
    369
    370
    371
    372
    373
    374
    375
    376
    377
    378
    379
    380
    381
    382
    383
    384
    385
    386
    387
    388
    389
    390
    391
    392
    393

    def encode_table_value(
        value: common.FieldArray | common.FieldTable | common.FieldValue,
    ) -> bytes:
        """Takes a value of any type and tries to encode it with the proper encoder

        :param value: Value to encode
        :type value: :const:`pamqp.common.FieldArray` or
                     :const:`pamqp.common.FieldTable` or
                     :const:`pamqp.common.FieldValue`
        :raises TypeError: when the type of the value is not supported

        """
        if isinstance(value, bool):
            return b't' + boolean(value)
        elif isinstance(value, int):
            return table_integer(value)
        elif isinstance(value, _decimal.Decimal):
            return b'D' + decimal(value)
        elif isinstance(value, float):
            return b'f' + floating_point(value)
        elif isinstance(value, str):
            return b'S' + long_string(value)
        elif isinstance(value, (datetime.datetime, time.struct_time)):
            return b'T' + timestamp(value)
        elif isinstance(value, dict):
            return b'F' + field_table(value)
        elif isinstance(value, list):
            return b'A' + field_array(value)
        elif isinstance(value, bytearray):
            return b'x' + byte_array(value)
        elif value is None:
            return b'V'
        raise TypeError(f'Unknown type: {type(value)} ({value!r})')