Skip to content

pamqp.header

pamqp.header

AMQP Header Class Definitions

For encoding AMQP Header frames into binary AMQP stream data and decoding AMQP binary data into AMQP Header frames.

ProtocolHeader

Class that represents the AMQP Protocol Header

Source code in pamqp/header.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class ProtocolHeader:
    """Class that represents the AMQP Protocol Header"""

    name: typing.ClassVar[str] = 'ProtocolHeader'

    def __init__(
        self,
        major_version: int = constants.VERSION[0],
        minor_version: int = constants.VERSION[1],
        revision: int = constants.VERSION[2],
    ) -> None:
        """Construct a Protocol Header frame object for the specified AMQP
        version.

        :param major_version: The AMQP major version (``0``)
        :param minor_version: The AMQP major version (``9``)
        :param revision: The AMQP major version (``1``)

        """
        self.major_version = major_version
        self.minor_version = minor_version
        self.revision = revision

    def marshal(self) -> bytes:
        """Return the full AMQP wire protocol frame data representation of the
        ProtocolHeader frame.

        """
        return constants.AMQP + struct.pack(
            'BBBB', 0, self.major_version, self.minor_version, self.revision
        )

    def unmarshal(self, data: bytes) -> int:
        """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 frame value to unpack
        :raises: ValueError

        """
        try:
            (self.major_version, self.minor_version, self.revision) = (
                struct.unpack('BBB', data[5:8])
            )
        except struct.error as err:
            raise ValueError(
                f'Could not unpack protocol header from {data!r}'
            ) from err
        return 8

__init__(major_version=constants.VERSION[0], minor_version=constants.VERSION[1], revision=constants.VERSION[2])

Construct a Protocol Header frame object for the specified AMQP version.

Parameters:

Name Type Description Default
major_version int

The AMQP major version (0)

 VERSION[0]
minor_version int

The AMQP major version (9)

 VERSION[1]
revision int

The AMQP major version (1)

 VERSION[2]
Source code in pamqp/header.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    major_version: int = constants.VERSION[0],
    minor_version: int = constants.VERSION[1],
    revision: int = constants.VERSION[2],
) -> None:
    """Construct a Protocol Header frame object for the specified AMQP
    version.

    :param major_version: The AMQP major version (``0``)
    :param minor_version: The AMQP major version (``9``)
    :param revision: The AMQP major version (``1``)

    """
    self.major_version = major_version
    self.minor_version = minor_version
    self.revision = revision

marshal()

Return the full AMQP wire protocol frame data representation of the ProtocolHeader frame.

Source code in pamqp/header.py 
40
41
42
43
44
45
46
47
def marshal(self) -> bytes:
    """Return the full AMQP wire protocol frame data representation of the
    ProtocolHeader frame.

    """
    return constants.AMQP + struct.pack(
        'BBBB', 0, self.major_version, self.minor_version, self.revision
    )

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 frame value to unpack

 required
Source code in pamqp/header.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def unmarshal(self, data: bytes) -> int:
    """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 frame value to unpack
    :raises: ValueError

    """
    try:
        (self.major_version, self.minor_version, self.revision) = (
            struct.unpack('BBB', data[5:8])
        )
    except struct.error as err:
        raise ValueError(
            f'Could not unpack protocol header from {data!r}'
        ) from err
    return 8

ContentHeader

Represent a content header frame

A Content Header frame is received after a Basic.Deliver or Basic.GetOk frame and has the data and properties for the Content Body frames that follow.

Source code in pamqp/header.py 
 68
 69
 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
class ContentHeader:
    """Represent a content header frame

    A Content Header frame is received after a Basic.Deliver or Basic.GetOk
    frame and has the data and properties for the Content Body frames that
    follow.

    """

    name: typing.ClassVar[str] = 'ContentHeader'

    def __init__(
        self,
        weight: int = 0,
        body_size: int = 0,
        properties: commands.Basic.Properties | None = None,
    ) -> None:
        """Initialize the Exchange.DeleteOk class

        Weight is unused and must be `0`

        :param weight: The unused content weight field
        :param body_size: The size in bytes of the message body
        :param properties: The message properties

        """
        self.class_id = None
        self.weight = weight
        self.body_size = body_size
        self.properties = properties or commands.Basic.Properties()

    def marshal(self) -> bytes:
        """Return the AMQP binary encoded value of the frame"""
        return (
            struct.pack('>HxxQ', commands.Basic.frame_id, self.body_size)
            + self.properties.marshal()
        )

    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 frame data to unmarshal

        """
        self.class_id, self.weight, self.body_size = struct.unpack(
            '>HHQ', data[0:12]
        )
        offset, flags = self._get_flags(data[12:])
        self.properties.unmarshal(flags, data[12 + offset :])

    @staticmethod
    def _get_flags(data: bytes) -> tuple[int, int]:
        """Decode the flags from the data returning the bytes consumed and
        flags.

        """
        bytes_consumed, flags, flagword_index = 0, 0, 0
        while True:
            consumed, partial_flags = decode.short_int(data)
            bytes_consumed += consumed
            flags |= partial_flags << (flagword_index * 16)
            if not partial_flags & 1:  # pragma: nocover
                break
            flagword_index += 1  # pragma: nocover
        return bytes_consumed, flags

__init__(weight=0, body_size=0, properties=None)

Initialize the Exchange.DeleteOk class

Weight is unused and must be 0

Parameters:

Name Type Description Default
weight int

The unused content weight field

 0
body_size int

The size in bytes of the message body

 0
properties Properties | None

The message properties

 None
Source code in pamqp/header.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    weight: int = 0,
    body_size: int = 0,
    properties: commands.Basic.Properties | None = None,
) -> None:
    """Initialize the Exchange.DeleteOk class

    Weight is unused and must be `0`

    :param weight: The unused content weight field
    :param body_size: The size in bytes of the message body
    :param properties: The message properties

    """
    self.class_id = None
    self.weight = weight
    self.body_size = body_size
    self.properties = properties or commands.Basic.Properties()

marshal()

Return the AMQP binary encoded value of the frame

Source code in pamqp/header.py 
 99
100
101
102
103
104
def marshal(self) -> bytes:
    """Return the AMQP binary encoded value of the frame"""
    return (
        struct.pack('>HxxQ', commands.Basic.frame_id, self.body_size)
        + self.properties.marshal()
    )

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 frame data to unmarshal

 required
Source code in pamqp/header.py 
106
107
108
109
110
111
112
113
114
115
116
117
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 frame data to unmarshal

    """
    self.class_id, self.weight, self.body_size = struct.unpack(
        '>HHQ', data[0:12]
    )
    offset, flags = self._get_flags(data[12:])
    self.properties.unmarshal(flags, data[12 + offset :])