Skip to content

# pamqp.base

## `pamqp.base`

Base classes for the representation of frames and data structures.

### `Frame`

Bases: `_AMQData`

Base Class for AMQ Methods for encoding and decoding

Source code in `pamqp/base.py`

     70
     71
     72
     73
     74
     75
     76
     77
     78
     79
     80
     81
     82
     83
     84
     85
     86
     87
     88
     89
     90
     91
     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139

    class Frame(_AMQData):
        """Base Class for AMQ Methods for encoding and decoding"""

        frame_id: typing.ClassVar[int] = 0
        index: typing.ClassVar[int] = 0
        synchronous: typing.ClassVar[bool] = False
        valid_responses: typing.ClassVar[list[str]] = []

        def marshal(self) -> bytes:
            """Dynamically encode the frame by taking the list of attributes and
            encode them item by item getting the value form the object attribute
            and the data type from the class attribute.

            """
            self.validate()
            byte, offset, output, processing_bitset = -1, 0, [], False
            for argument in self.__slots__:
                data_type = self.amqp_type(argument)
                if not processing_bitset and data_type == 'bit':
                    byte, offset, processing_bitset = 0, 0, True
                data_value = getattr(self, argument, 0)
                if processing_bitset:
                    if data_type != 'bit':
                        processing_bitset = False
                        output.append(encode.octet(byte))
                    else:
                        byte = encode.bit(data_value, byte, offset)
                        offset += 1
                        if offset == 8:  # pragma: nocover
                            output.append(encode.octet(byte))
                            processing_bitset = False
                        continue  # pragma: nocover
                output.append(encode.by_type(data_value, data_type))
            if processing_bitset:
                output.append(encode.octet(byte))
            return b''.join(output)

        def unmarshal(self, data: bytes) -> None:
            """Dynamically decode the frame data applying the values to the method
            object by iterating through the attributes in order and decoding them.

            :param data: The raw AMQP frame data

            """
            offset, processing_bitset = 0, False
            for argument in self.__slots__:
                data_type = self.amqp_type(argument)
                if offset == 7 and processing_bitset:  # pragma: nocover
                    data = data[1:]
                    offset = 0
                if processing_bitset and data_type != 'bit':
                    offset = 0
                    processing_bitset = False
                    data = data[1:]
                consumed, value = decode.by_type(data, data_type, offset)
                if data_type == 'bit':
                    offset += 1
                    processing_bitset = True
                    consumed = 0
                setattr(self, argument, value)
                if consumed:
                    data = data[consumed:]

        def validate(self) -> None:
            """Validate the frame data ensuring all domains or attributes adhere
            to the protocol specification.

            :raises: ValueError

            """

#### `marshal()`

Dynamically encode the frame by taking the list of attributes and encode them item by item getting the value form the object attribute and the data type from the class attribute.

Source code in `pamqp/base.py`

     78
     79
     80
     81
     82
     83
     84
     85
     86
     87
     88
     89
     90
     91
     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105

    def marshal(self) -> bytes:
        """Dynamically encode the frame by taking the list of attributes and
        encode them item by item getting the value form the object attribute
        and the data type from the class attribute.

        """
        self.validate()
        byte, offset, output, processing_bitset = -1, 0, [], False
        for argument in self.__slots__:
            data_type = self.amqp_type(argument)
            if not processing_bitset and data_type == 'bit':
                byte, offset, processing_bitset = 0, 0, True
            data_value = getattr(self, argument, 0)
            if processing_bitset:
                if data_type != 'bit':
                    processing_bitset = False
                    output.append(encode.octet(byte))
                else:
                    byte = encode.bit(data_value, byte, offset)
                    offset += 1
                    if offset == 8:  # pragma: nocover
                        output.append(encode.octet(byte))
                        processing_bitset = False
                    continue  # pragma: nocover
            output.append(encode.by_type(data_value, data_type))
        if processing_bitset:
            output.append(encode.octet(byte))
        return b''.join(output)

#### `unmarshal(data)`

Dynamically decode the frame data applying the values to the method object by iterating through the attributes in order and decoding them.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`data` | `bytes` | The raw AMQP frame data | required

Source code in `pamqp/base.py`

    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131

    def unmarshal(self, data: bytes) -> None:
        """Dynamically decode the frame data applying the values to the method
        object by iterating through the attributes in order and decoding them.

        :param data: The raw AMQP frame data

        """
        offset, processing_bitset = 0, False
        for argument in self.__slots__:
            data_type = self.amqp_type(argument)
            if offset == 7 and processing_bitset:  # pragma: nocover
                data = data[1:]
                offset = 0
            if processing_bitset and data_type != 'bit':
                offset = 0
                processing_bitset = False
                data = data[1:]
            consumed, value = decode.by_type(data, data_type, offset)
            if data_type == 'bit':
                offset += 1
                processing_bitset = True
                consumed = 0
            setattr(self, argument, value)
            if consumed:
                data = data[consumed:]

#### `validate()`

Validate the frame data ensuring all domains or attributes adhere to the protocol specification.

Source code in `pamqp/base.py`

    133
    134
    135
    136
    137
    138
    139

    def validate(self) -> None:
        """Validate the frame data ensuring all domains or attributes adhere
        to the protocol specification.

        :raises: ValueError

        """

### `BasicProperties`

Bases: `_AMQData`

Provide a base object that marshals and unmarshals the Basic.Properties object values.

Source code in `pamqp/base.py`

    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219

    class BasicProperties(_AMQData):
        """Provide a base object that marshals and unmarshals the Basic.Properties
        object values.

        """

        flags: typing.ClassVar[dict[str, int]] = {}
        name: typing.ClassVar[str] = 'BasicProperties'

        def __eq__(self, other: object) -> bool:
            if not isinstance(other, BasicProperties):
                raise NotImplementedError
            return all(
                getattr(self, k, None) == getattr(other, k, None)
                for k in self.__slots__
            )

        def encode_property(self, name: str, value: common.FieldValue) -> bytes:
            """Encode a single property value

            :param name: The name of the property to encode
            :param value: The property to encode
            :type value: :const:`pamqp.common.FieldValue`
            :raises: TypeError

            """
            return encode.by_type(value, self.amqp_type(name))

        def marshal(self) -> bytes:
            """Take the Basic.Properties data structure and marshal it into the
            data structure needed for the ContentHeader.

            """
            flags = 0
            parts = []
            for property_name in self.__slots__:
                property_value = getattr(self, property_name)
                if property_value is not None and property_value != '':
                    flags = flags | self.flags[property_name]
                    parts.append(
                        self.encode_property(property_name, property_value)
                    )
            flag_pieces = []
            while True:
                remainder = flags >> 16
                partial_flags = flags & 0xFFFE
                if remainder != 0:  # pragma: nocover
                    partial_flags |= 1
                flag_pieces.append(struct.pack('>H', partial_flags))
                flags = remainder
                if not flags:  # pragma: nocover
                    break
            return b''.join(flag_pieces + parts)

        def unmarshal(self, flags: int, data: bytes) -> None:
            """Dynamically decode the frame data applying the values to the method
            object by iterating through the attributes in order and decoding them.

            """
            for property_name in self.__slots__:
                if flags & self.flags[property_name]:
                    data_type = getattr(self.__class__, '_' + property_name)
                    consumed, value = decode.by_type(data, data_type)
                    setattr(self, property_name, value)
                    data = data[consumed:]

        def validate(self) -> None:
            """Validate the frame data ensuring all domains or attributes adhere
            to the protocol specification.

            :raises: ValueError

            """
            if getattr(self, 'cluster_id', '') != '':
                raise ValueError('cluster_id must be empty')
            delivery_mode = getattr(self, 'delivery_mode', None)
            if delivery_mode is not None and delivery_mode not in [1, 2]:
                raise ValueError(f'Invalid delivery_mode value: {delivery_mode}')

#### `encode_property(name, value)`

Encode a single property value

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`name` | `str` | The name of the property to encode | required
`value` | `FieldValue` | The property to encode | required

Source code in `pamqp/base.py`

    159
    160
    161
    162
    163
    164
    165
    166
    167
    168

    def encode_property(self, name: str, value: common.FieldValue) -> bytes:
        """Encode a single property value

        :param name: The name of the property to encode
        :param value: The property to encode
        :type value: :const:`pamqp.common.FieldValue`
        :raises: TypeError

        """
        return encode.by_type(value, self.amqp_type(name))

#### `marshal()`

Take the Basic.Properties data structure and marshal it into the data structure needed for the ContentHeader.

Source code in `pamqp/base.py`

    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194

    def marshal(self) -> bytes:
        """Take the Basic.Properties data structure and marshal it into the
        data structure needed for the ContentHeader.

        """
        flags = 0
        parts = []
        for property_name in self.__slots__:
            property_value = getattr(self, property_name)
            if property_value is not None and property_value != '':
                flags = flags | self.flags[property_name]
                parts.append(
                    self.encode_property(property_name, property_value)
                )
        flag_pieces = []
        while True:
            remainder = flags >> 16
            partial_flags = flags & 0xFFFE
            if remainder != 0:  # pragma: nocover
                partial_flags |= 1
            flag_pieces.append(struct.pack('>H', partial_flags))
            flags = remainder
            if not flags:  # pragma: nocover
                break
        return b''.join(flag_pieces + parts)

#### `unmarshal(flags, data)`

Dynamically decode the frame data applying the values to the method object by iterating through the attributes in order and decoding them.

Source code in `pamqp/base.py`

    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206

    def unmarshal(self, flags: int, data: bytes) -> None:
        """Dynamically decode the frame data applying the values to the method
        object by iterating through the attributes in order and decoding them.

        """
        for property_name in self.__slots__:
            if flags & self.flags[property_name]:
                data_type = getattr(self.__class__, '_' + property_name)
                consumed, value = decode.by_type(data, data_type)
                setattr(self, property_name, value)
                data = data[consumed:]

#### `validate()`

Validate the frame data ensuring all domains or attributes adhere to the protocol specification.

Source code in `pamqp/base.py`

    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219

    def validate(self) -> None:
        """Validate the frame data ensuring all domains or attributes adhere
        to the protocol specification.

        :raises: ValueError

        """
        if getattr(self, 'cluster_id', '') != '':
            raise ValueError('cluster_id must be empty')
        delivery_mode = getattr(self, 'delivery_mode', None)
        if delivery_mode is not None and delivery_mode not in [1, 2]:
            raise ValueError(f'Invalid delivery_mode value: {delivery_mode}')