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}')