pamqp.commands

pamqp.commands

The classes inside :mod:pamqp.commands allow for the automatic marshaling and unmarshaling of AMQP method frames and :class:Basic.Properties <pamqp.commands.Basic.Properties>. In addition the command classes contain information that designates if they are synchronous commands and if so, what the expected responses are. Each commands arguments are detailed in the class and are listed in the attributes property.

.. note:: All AMQ classes and methods extend :class:pamqp.base.Frame.

Connection

Work with socket connections

The connection class provides methods for a client to establish a network connection to a server, and for both peers to operate the connection thereafter.

Source code in pamqp/commands.py

class Connection:
    """Work with socket connections

    The connection class provides methods for a client to establish a network
    connection to a server, and for both peers to operate the connection
    thereafter.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0000  # pamqp Mapping Index

    class Start(base.Frame):
        """Start connection negotiation

        This method starts the connection negotiation process by telling the
        client the protocol version that the server proposes, along with a list
        of security mechanisms which the client can use for authentication.

        :param version_major: Protocol major version
            - Default: ``0``
        :param version_minor: Protocol minor version
            - Default: ``9``
        :param server_properties: Server properties
            - Default: ``{}``
        :type server_properties: :const:`~pamqp.common.FieldTable`
        :param mechanisms: Available security mechanisms
            - Default: ``PLAIN``
        :param locales: Available message locales
            - Default: ``en_US``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'version_major', 'version_minor', 'server_properties',
            'mechanisms', 'locales'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Start'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Connection.StartOk']

        # Class Attribute Types for unmarshaling
        _version_major = 'octet'
        _version_minor = 'octet'
        _server_properties = 'table'
        _mechanisms = 'longstr'
        _locales = 'longstr'

        def __init__(self,
                     version_major: int = 0,
                     version_minor: int = 9,
                     server_properties: common.FieldTable | None = None,
                     mechanisms: str = 'PLAIN',
                     locales: str = 'en_US') -> None:
            """Initialize the :class:`Connection.Start` class"""
            self.version_major = version_major
            self.version_minor = version_minor
            self.server_properties = server_properties or {}
            self.mechanisms = mechanisms
            self.locales = locales

    class StartOk(base.Frame):
        """Select security mechanism and locale

        This method selects a SASL security mechanism.

        :param client_properties: Client properties
            - Default: ``{}``
        :type client_properties: :const:`~pamqp.common.FieldTable`
        :param mechanism: Selected security mechanism
            - Default: ``PLAIN``
        :param response: Security response data
            - Default: ``''``
        :param locale: Selected message locale
            - Default: ``en_US``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'client_properties', 'mechanism', 'response', 'locale'
        ]

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.StartOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _client_properties = 'table'
        _mechanism = 'shortstr'
        _response = 'longstr'
        _locale = 'shortstr'

        def __init__(self,
                     client_properties: common.FieldTable | None = None,
                     mechanism: str = 'PLAIN',
                     response: str = '',
                     locale: str = 'en_US') -> None:
            """Initialize the :class:`Connection.StartOk` class"""
            self.client_properties = client_properties or {}
            self.mechanism = mechanism
            self.response = response
            self.locale = locale

    class Secure(base.Frame):
        """Security mechanism challenge

        The SASL protocol works by exchanging challenges and responses until
        both peers have received sufficient information to authenticate each
        other. This method challenges the client to provide more information.

        :param challenge: Security challenge data

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'challenge'
        ]

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Secure'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Connection.SecureOk']

        # Class Attribute Types for unmarshaling
        _challenge = 'longstr'

        def __init__(self, challenge: str | None = None) -> None:
            """Initialize the :class:`Connection.Secure` class"""
            self.challenge = challenge

    class SecureOk(base.Frame):
        """Security mechanism response

        This method attempts to authenticate, passing a block of SASL data for
        the security mechanism at the server side.

        :param response: Security response data

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'response'
        ]

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.SecureOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _response = 'longstr'

        def __init__(self, response: str | None = None) -> None:
            """Initialize the :class:`Connection.SecureOk` class"""
            self.response = response

    class Tune(base.Frame):
        """Propose connection tuning parameters

        This method proposes a set of connection configuration values to the
        client. The client can accept and/or adjust these.

        :param channel_max: Proposed maximum channels
            - Default: ``0``
        :param frame_max: Proposed maximum frame size
            - Default: ``0``
        :param heartbeat: Desired heartbeat delay
            - Default: ``0``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'channel_max', 'frame_max', 'heartbeat'
        ]

        frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A001E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Tune'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Connection.TuneOk']

        # Class Attribute Types for unmarshaling
        _channel_max = 'short'
        _frame_max = 'long'
        _heartbeat = 'short'

        def __init__(self,
                     channel_max: int = 0,
                     frame_max: int = 0,
                     heartbeat: int = 0) -> None:
            """Initialize the :class:`Connection.Tune` class"""
            self.channel_max = channel_max
            self.frame_max = frame_max
            self.heartbeat = heartbeat

    class TuneOk(base.Frame):
        """Negotiate connection tuning parameters

        This method sends the client's connection tuning parameters to the
        server. Certain fields are negotiated, others provide capability
        information.

        :param channel_max: Negotiated maximum channels
            - Default: ``0``
        :param frame_max: Negotiated maximum frame size
            - Default: ``0``
        :param heartbeat: Desired heartbeat delay
            - Default: ``0``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'channel_max', 'frame_max', 'heartbeat'
        ]

        frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A001F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.TuneOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _channel_max = 'short'
        _frame_max = 'long'
        _heartbeat = 'short'

        def __init__(self,
                     channel_max: int = 0,
                     frame_max: int = 0,
                     heartbeat: int = 0) -> None:
            """Initialize the :class:`Connection.TuneOk` class"""
            self.channel_max = channel_max
            self.frame_max = frame_max
            self.heartbeat = heartbeat

    class Open(base.Frame):
        """Open connection to virtual host

        This method opens a connection to a virtual host, which is a collection
        of resources, and acts to separate multiple application domains within
        a server. The server may apply arbitrary limits per virtual host, such
        as the number of each type of entity that may be used, per connection
        and/or in total.

        :param virtual_host: Virtual host name
            - Default: ``/``
        :param capabilities: Deprecated, must be empty
            - Default: ``''``
        :param insist: Deprecated, must be ``False``
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'virtual_host', 'capabilities', 'insist'
        ]

        frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0028  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Open'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Connection.OpenOk']

        # Class Attribute Types for unmarshaling
        _virtual_host = 'shortstr'
        _capabilities = 'shortstr'
        _insist = 'bit'

        def __init__(self,
                     virtual_host: str = '/',
                     capabilities: str = '',
                     insist: bool = False) -> None:
            """Initialize the :class:`Connection.Open` class"""
            self.virtual_host = virtual_host
            self.capabilities = capabilities
            self.insist = insist
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.virtual_host is not None and len(self.virtual_host) > 127:
                raise ValueError('Max length exceeded for virtual_host')
            if self.capabilities is not None and self.capabilities != '':
                raise ValueError('capabilities must be empty')
            if self.insist is not None and self.insist is not False:
                raise ValueError('insist must be False')

    class OpenOk(base.Frame):
        """Signal that connection is ready

        This method signals to the client that the connection is ready for use.

        :param known_hosts: Deprecated, must be empty
            - Default: ``''``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'known_hosts'
        ]

        frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0029  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.OpenOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _known_hosts = 'shortstr'

        def __init__(self, known_hosts: str = '') -> None:
            """Initialize the :class:`Connection.OpenOk` class"""
            self.known_hosts = known_hosts
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.known_hosts is not None and self.known_hosts != '':
                raise ValueError('known_hosts must be empty')

    class Close(base.Frame):
        """Request a connection close

        This method indicates that the sender wants to close the connection.
        This may be due to internal conditions (e.g. a forced shut-down) or due
        to an error handling a specific method, i.e. an exception. When a close
        is due to an exception, the sender provides the class and method id of
        the method which caused the exception.

        :param reply_code: Reply code from server
        :param reply_text: Localised reply text
            - Default: ``''``
        :param class_id: Failing method class
        :param method_id: Failing method ID

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'reply_code', 'reply_text', 'class_id', 'method_id'
        ]

        frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0032  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Close'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Connection.CloseOk']

        # Class Attribute Types for unmarshaling
        _reply_code = 'short'
        _reply_text = 'shortstr'
        _class_id = 'short'
        _method_id = 'short'

        def __init__(self,
                     reply_code: int | None = None,
                     reply_text: str = '',
                     class_id: int | None = None,
                     method_id: int | None = None) -> None:
            """Initialize the :class:`Connection.Close` class"""
            self.reply_code = reply_code
            self.reply_text = reply_text or ''
            self.class_id = class_id
            self.method_id = method_id

    class CloseOk(base.Frame):
        """Confirm a connection close

        This method confirms a :class:`Connection.Close` method and tells the
        recipient that it is safe to release resources for the connection and
        close the socket.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0033  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.CloseOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Blocked(base.Frame):
        """Indicate that connection is blocked

        This method indicates that a connection has been blocked and does not
        accept new publishes.

        :param reason: Block reason
            - Default: ``''``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'reason'
        ]

        frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A003C  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Blocked'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _reason = 'shortstr'

        def __init__(self, reason: str = '') -> None:
            """Initialize the :class:`Connection.Blocked` class"""
            self.reason = reason

    class Unblocked(base.Frame):
        """Indicate that connection is unblocked

        This method indicates that a connection has been unblocked and now
        accepts publishes.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 61  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A003D  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.Unblocked'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class UpdateSecret(base.Frame):
        """Update secret

        This method updates the secret used to authenticate this connection. It
        is used when secrets have an expiration date and need to be renewed,
        like OAuth 2 tokens.

        :param new_secret: New secret
        :param reason: Reason

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'new_secret', 'reason'
        ]

        frame_id: typing.ClassVar[int] = 70  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0046  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.UpdateSecret'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = [
            'Connection.UpdateSecretOk'
        ]

        # Class Attribute Types for unmarshaling
        _new_secret = 'longstr'
        _reason = 'shortstr'

        def __init__(self,
                     new_secret: str | None = None,
                     reason: str | None = None) -> None:
            """Initialize the :class:`Connection.UpdateSecret` class"""
            self.new_secret = new_secret
            self.reason = reason

    class UpdateSecretOk(base.Frame):
        """Update secret response

        This method confirms the updated secret is valid.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 71  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x000A0047  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Connection.UpdateSecretOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Start

Bases: Frame

Start connection negotiation

This method starts the connection negotiation process by telling the client the protocol version that the server proposes, along with a list of security mechanisms which the client can use for authentication.

Parameters:

Name	Type	Description	Default
version_major	int	Protocol major version - Default: 0	0
version_minor	int	Protocol minor version - Default: 9	9
server_properties	FieldTable | None	Server properties - Default: {}	None
mechanisms	str	Available security mechanisms - Default: PLAIN	'PLAIN'
locales	str	Available message locales - Default: en_US	'en_US'

Source code in pamqp/commands.py

class Start(base.Frame):
    """Start connection negotiation

    This method starts the connection negotiation process by telling the
    client the protocol version that the server proposes, along with a list
    of security mechanisms which the client can use for authentication.

    :param version_major: Protocol major version
        - Default: ``0``
    :param version_minor: Protocol minor version
        - Default: ``9``
    :param server_properties: Server properties
        - Default: ``{}``
    :type server_properties: :const:`~pamqp.common.FieldTable`
    :param mechanisms: Available security mechanisms
        - Default: ``PLAIN``
    :param locales: Available message locales
        - Default: ``en_US``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'version_major', 'version_minor', 'server_properties',
        'mechanisms', 'locales'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Start'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Connection.StartOk']

    # Class Attribute Types for unmarshaling
    _version_major = 'octet'
    _version_minor = 'octet'
    _server_properties = 'table'
    _mechanisms = 'longstr'
    _locales = 'longstr'

    def __init__(self,
                 version_major: int = 0,
                 version_minor: int = 9,
                 server_properties: common.FieldTable | None = None,
                 mechanisms: str = 'PLAIN',
                 locales: str = 'en_US') -> None:
        """Initialize the :class:`Connection.Start` class"""
        self.version_major = version_major
        self.version_minor = version_minor
        self.server_properties = server_properties or {}
        self.mechanisms = mechanisms
        self.locales = locales

__init__(version_major=0, version_minor=9, server_properties=None, mechanisms='PLAIN', locales='en_US')

Initialize the :class:Connection.Start class

Source code in pamqp/commands.py

def __init__(self,
             version_major: int = 0,
             version_minor: int = 9,
             server_properties: common.FieldTable | None = None,
             mechanisms: str = 'PLAIN',
             locales: str = 'en_US') -> None:
    """Initialize the :class:`Connection.Start` class"""
    self.version_major = version_major
    self.version_minor = version_minor
    self.server_properties = server_properties or {}
    self.mechanisms = mechanisms
    self.locales = locales

StartOk

Bases: Frame

Select security mechanism and locale

This method selects a SASL security mechanism.

Parameters:

Name	Type	Description	Default
client_properties	FieldTable | None	Client properties - Default: {}	None
mechanism	str	Selected security mechanism - Default: PLAIN	'PLAIN'
response	str	Security response data - Default: ''	''
locale	str	Selected message locale - Default: en_US	'en_US'

Source code in pamqp/commands.py

class StartOk(base.Frame):
    """Select security mechanism and locale

    This method selects a SASL security mechanism.

    :param client_properties: Client properties
        - Default: ``{}``
    :type client_properties: :const:`~pamqp.common.FieldTable`
    :param mechanism: Selected security mechanism
        - Default: ``PLAIN``
    :param response: Security response data
        - Default: ``''``
    :param locale: Selected message locale
        - Default: ``en_US``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'client_properties', 'mechanism', 'response', 'locale'
    ]

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.StartOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _client_properties = 'table'
    _mechanism = 'shortstr'
    _response = 'longstr'
    _locale = 'shortstr'

    def __init__(self,
                 client_properties: common.FieldTable | None = None,
                 mechanism: str = 'PLAIN',
                 response: str = '',
                 locale: str = 'en_US') -> None:
        """Initialize the :class:`Connection.StartOk` class"""
        self.client_properties = client_properties or {}
        self.mechanism = mechanism
        self.response = response
        self.locale = locale

__init__(client_properties=None, mechanism='PLAIN', response='', locale='en_US')

Initialize the :class:Connection.StartOk class

Source code in pamqp/commands.py

def __init__(self,
             client_properties: common.FieldTable | None = None,
             mechanism: str = 'PLAIN',
             response: str = '',
             locale: str = 'en_US') -> None:
    """Initialize the :class:`Connection.StartOk` class"""
    self.client_properties = client_properties or {}
    self.mechanism = mechanism
    self.response = response
    self.locale = locale

Secure

Bases: Frame

Security mechanism challenge

The SASL protocol works by exchanging challenges and responses until both peers have received sufficient information to authenticate each other. This method challenges the client to provide more information.

Parameters:

Name	Type	Description	Default
challenge	str | None	Security challenge data	None

Source code in pamqp/commands.py

class Secure(base.Frame):
    """Security mechanism challenge

    The SASL protocol works by exchanging challenges and responses until
    both peers have received sufficient information to authenticate each
    other. This method challenges the client to provide more information.

    :param challenge: Security challenge data

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'challenge'
    ]

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Secure'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Connection.SecureOk']

    # Class Attribute Types for unmarshaling
    _challenge = 'longstr'

    def __init__(self, challenge: str | None = None) -> None:
        """Initialize the :class:`Connection.Secure` class"""
        self.challenge = challenge

__init__(challenge=None)

Initialize the :class:Connection.Secure class

Source code in pamqp/commands.py

def __init__(self, challenge: str | None = None) -> None:
    """Initialize the :class:`Connection.Secure` class"""
    self.challenge = challenge

SecureOk

Bases: Frame

Security mechanism response

This method attempts to authenticate, passing a block of SASL data for the security mechanism at the server side.

Parameters:

Name	Type	Description	Default
response	str | None	Security response data	None

Source code in pamqp/commands.py

class SecureOk(base.Frame):
    """Security mechanism response

    This method attempts to authenticate, passing a block of SASL data for
    the security mechanism at the server side.

    :param response: Security response data

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'response'
    ]

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.SecureOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _response = 'longstr'

    def __init__(self, response: str | None = None) -> None:
        """Initialize the :class:`Connection.SecureOk` class"""
        self.response = response

__init__(response=None)

Initialize the :class:Connection.SecureOk class

Source code in pamqp/commands.py

def __init__(self, response: str | None = None) -> None:
    """Initialize the :class:`Connection.SecureOk` class"""
    self.response = response

Tune

Bases: Frame

Propose connection tuning parameters

This method proposes a set of connection configuration values to the client. The client can accept and/or adjust these.

Parameters:

Name	Type	Description	Default
channel_max	int	Proposed maximum channels - Default: 0	0
frame_max	int	Proposed maximum frame size - Default: 0	0
heartbeat	int	Desired heartbeat delay - Default: 0	0

Source code in pamqp/commands.py

class Tune(base.Frame):
    """Propose connection tuning parameters

    This method proposes a set of connection configuration values to the
    client. The client can accept and/or adjust these.

    :param channel_max: Proposed maximum channels
        - Default: ``0``
    :param frame_max: Proposed maximum frame size
        - Default: ``0``
    :param heartbeat: Desired heartbeat delay
        - Default: ``0``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'channel_max', 'frame_max', 'heartbeat'
    ]

    frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A001E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Tune'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Connection.TuneOk']

    # Class Attribute Types for unmarshaling
    _channel_max = 'short'
    _frame_max = 'long'
    _heartbeat = 'short'

    def __init__(self,
                 channel_max: int = 0,
                 frame_max: int = 0,
                 heartbeat: int = 0) -> None:
        """Initialize the :class:`Connection.Tune` class"""
        self.channel_max = channel_max
        self.frame_max = frame_max
        self.heartbeat = heartbeat

__init__(channel_max=0, frame_max=0, heartbeat=0)

Initialize the :class:Connection.Tune class

Source code in pamqp/commands.py

def __init__(self,
             channel_max: int = 0,
             frame_max: int = 0,
             heartbeat: int = 0) -> None:
    """Initialize the :class:`Connection.Tune` class"""
    self.channel_max = channel_max
    self.frame_max = frame_max
    self.heartbeat = heartbeat

TuneOk

Bases: Frame

Negotiate connection tuning parameters

This method sends the client's connection tuning parameters to the server. Certain fields are negotiated, others provide capability information.

Parameters:

Name	Type	Description	Default
channel_max	int	Negotiated maximum channels - Default: 0	0
frame_max	int	Negotiated maximum frame size - Default: 0	0
heartbeat	int	Desired heartbeat delay - Default: 0	0

Source code in pamqp/commands.py

class TuneOk(base.Frame):
    """Negotiate connection tuning parameters

    This method sends the client's connection tuning parameters to the
    server. Certain fields are negotiated, others provide capability
    information.

    :param channel_max: Negotiated maximum channels
        - Default: ``0``
    :param frame_max: Negotiated maximum frame size
        - Default: ``0``
    :param heartbeat: Desired heartbeat delay
        - Default: ``0``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'channel_max', 'frame_max', 'heartbeat'
    ]

    frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A001F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.TuneOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _channel_max = 'short'
    _frame_max = 'long'
    _heartbeat = 'short'

    def __init__(self,
                 channel_max: int = 0,
                 frame_max: int = 0,
                 heartbeat: int = 0) -> None:
        """Initialize the :class:`Connection.TuneOk` class"""
        self.channel_max = channel_max
        self.frame_max = frame_max
        self.heartbeat = heartbeat

__init__(channel_max=0, frame_max=0, heartbeat=0)

Initialize the :class:Connection.TuneOk class

Source code in pamqp/commands.py

def __init__(self,
             channel_max: int = 0,
             frame_max: int = 0,
             heartbeat: int = 0) -> None:
    """Initialize the :class:`Connection.TuneOk` class"""
    self.channel_max = channel_max
    self.frame_max = frame_max
    self.heartbeat = heartbeat

Open

Bases: Frame

Open connection to virtual host

This method opens a connection to a virtual host, which is a collection of resources, and acts to separate multiple application domains within a server. The server may apply arbitrary limits per virtual host, such as the number of each type of entity that may be used, per connection and/or in total.

Parameters:

Name	Type	Description	Default
virtual_host	str	Virtual host name - Default: /	'/'
capabilities	str	Deprecated, must be empty - Default: ''	''
insist	bool	Deprecated, must be False - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Open(base.Frame):
    """Open connection to virtual host

    This method opens a connection to a virtual host, which is a collection
    of resources, and acts to separate multiple application domains within
    a server. The server may apply arbitrary limits per virtual host, such
    as the number of each type of entity that may be used, per connection
    and/or in total.

    :param virtual_host: Virtual host name
        - Default: ``/``
    :param capabilities: Deprecated, must be empty
        - Default: ``''``
    :param insist: Deprecated, must be ``False``
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'virtual_host', 'capabilities', 'insist'
    ]

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0028  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Open'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Connection.OpenOk']

    # Class Attribute Types for unmarshaling
    _virtual_host = 'shortstr'
    _capabilities = 'shortstr'
    _insist = 'bit'

    def __init__(self,
                 virtual_host: str = '/',
                 capabilities: str = '',
                 insist: bool = False) -> None:
        """Initialize the :class:`Connection.Open` class"""
        self.virtual_host = virtual_host
        self.capabilities = capabilities
        self.insist = insist
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.virtual_host is not None and len(self.virtual_host) > 127:
            raise ValueError('Max length exceeded for virtual_host')
        if self.capabilities is not None and self.capabilities != '':
            raise ValueError('capabilities must be empty')
        if self.insist is not None and self.insist is not False:
            raise ValueError('insist must be False')

__init__(virtual_host='/', capabilities='', insist=False)

Initialize the :class:Connection.Open class

Source code in pamqp/commands.py

def __init__(self,
             virtual_host: str = '/',
             capabilities: str = '',
             insist: bool = False) -> None:
    """Initialize the :class:`Connection.Open` class"""
    self.virtual_host = virtual_host
    self.capabilities = capabilities
    self.insist = insist
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.virtual_host is not None and len(self.virtual_host) > 127:
        raise ValueError('Max length exceeded for virtual_host')
    if self.capabilities is not None and self.capabilities != '':
        raise ValueError('capabilities must be empty')
    if self.insist is not None and self.insist is not False:
        raise ValueError('insist must be False')

OpenOk

Bases: Frame

Signal that connection is ready

This method signals to the client that the connection is ready for use.

Parameters:

Name	Type	Description	Default
known_hosts	str	Deprecated, must be empty - Default: ''	''

Source code in pamqp/commands.py

class OpenOk(base.Frame):
    """Signal that connection is ready

    This method signals to the client that the connection is ready for use.

    :param known_hosts: Deprecated, must be empty
        - Default: ``''``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'known_hosts'
    ]

    frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0029  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.OpenOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _known_hosts = 'shortstr'

    def __init__(self, known_hosts: str = '') -> None:
        """Initialize the :class:`Connection.OpenOk` class"""
        self.known_hosts = known_hosts
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.known_hosts is not None and self.known_hosts != '':
            raise ValueError('known_hosts must be empty')

__init__(known_hosts='')

Initialize the :class:Connection.OpenOk class

Source code in pamqp/commands.py

def __init__(self, known_hosts: str = '') -> None:
    """Initialize the :class:`Connection.OpenOk` class"""
    self.known_hosts = known_hosts
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.known_hosts is not None and self.known_hosts != '':
        raise ValueError('known_hosts must be empty')

Close

Bases: Frame

Request a connection close

This method indicates that the sender wants to close the connection. This may be due to internal conditions (e.g. a forced shut-down) or due to an error handling a specific method, i.e. an exception. When a close is due to an exception, the sender provides the class and method id of the method which caused the exception.

Parameters:

Name	Type	Description	Default
reply_code	int | None	Reply code from server	None
reply_text	str	Localised reply text - Default: ''	''
class_id	int | None	Failing method class	None
method_id	int | None	Failing method ID	None

Source code in pamqp/commands.py

class Close(base.Frame):
    """Request a connection close

    This method indicates that the sender wants to close the connection.
    This may be due to internal conditions (e.g. a forced shut-down) or due
    to an error handling a specific method, i.e. an exception. When a close
    is due to an exception, the sender provides the class and method id of
    the method which caused the exception.

    :param reply_code: Reply code from server
    :param reply_text: Localised reply text
        - Default: ``''``
    :param class_id: Failing method class
    :param method_id: Failing method ID

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'reply_code', 'reply_text', 'class_id', 'method_id'
    ]

    frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0032  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Close'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Connection.CloseOk']

    # Class Attribute Types for unmarshaling
    _reply_code = 'short'
    _reply_text = 'shortstr'
    _class_id = 'short'
    _method_id = 'short'

    def __init__(self,
                 reply_code: int | None = None,
                 reply_text: str = '',
                 class_id: int | None = None,
                 method_id: int | None = None) -> None:
        """Initialize the :class:`Connection.Close` class"""
        self.reply_code = reply_code
        self.reply_text = reply_text or ''
        self.class_id = class_id
        self.method_id = method_id

__init__(reply_code=None, reply_text='', class_id=None, method_id=None)

Initialize the :class:Connection.Close class

Source code in pamqp/commands.py

def __init__(self,
             reply_code: int | None = None,
             reply_text: str = '',
             class_id: int | None = None,
             method_id: int | None = None) -> None:
    """Initialize the :class:`Connection.Close` class"""
    self.reply_code = reply_code
    self.reply_text = reply_text or ''
    self.class_id = class_id
    self.method_id = method_id

CloseOk

Bases: Frame

Confirm a connection close

This method confirms a :class:Connection.Close method and tells the recipient that it is safe to release resources for the connection and close the socket.

Source code in pamqp/commands.py

class CloseOk(base.Frame):
    """Confirm a connection close

    This method confirms a :class:`Connection.Close` method and tells the
    recipient that it is safe to release resources for the connection and
    close the socket.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0033  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.CloseOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Blocked

Bases: Frame

Indicate that connection is blocked

This method indicates that a connection has been blocked and does not accept new publishes.

Parameters:

Name	Type	Description	Default
reason	str	Block reason - Default: ''	''

Source code in pamqp/commands.py

class Blocked(base.Frame):
    """Indicate that connection is blocked

    This method indicates that a connection has been blocked and does not
    accept new publishes.

    :param reason: Block reason
        - Default: ``''``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'reason'
    ]

    frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A003C  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Blocked'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _reason = 'shortstr'

    def __init__(self, reason: str = '') -> None:
        """Initialize the :class:`Connection.Blocked` class"""
        self.reason = reason

__init__(reason='')

Initialize the :class:Connection.Blocked class

Source code in pamqp/commands.py

def __init__(self, reason: str = '') -> None:
    """Initialize the :class:`Connection.Blocked` class"""
    self.reason = reason

Unblocked

Bases: Frame

Indicate that connection is unblocked

This method indicates that a connection has been unblocked and now accepts publishes.

Source code in pamqp/commands.py

class Unblocked(base.Frame):
    """Indicate that connection is unblocked

    This method indicates that a connection has been unblocked and now
    accepts publishes.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 61  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A003D  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.Unblocked'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

UpdateSecret

Bases: Frame

Update secret

This method updates the secret used to authenticate this connection. It is used when secrets have an expiration date and need to be renewed, like OAuth 2 tokens.

Parameters:

Name	Type	Description	Default
new_secret	str | None	New secret	None
reason	str | None	Reason	None

Source code in pamqp/commands.py

class UpdateSecret(base.Frame):
    """Update secret

    This method updates the secret used to authenticate this connection. It
    is used when secrets have an expiration date and need to be renewed,
    like OAuth 2 tokens.

    :param new_secret: New secret
    :param reason: Reason

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'new_secret', 'reason'
    ]

    frame_id: typing.ClassVar[int] = 70  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0046  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.UpdateSecret'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = [
        'Connection.UpdateSecretOk'
    ]

    # Class Attribute Types for unmarshaling
    _new_secret = 'longstr'
    _reason = 'shortstr'

    def __init__(self,
                 new_secret: str | None = None,
                 reason: str | None = None) -> None:
        """Initialize the :class:`Connection.UpdateSecret` class"""
        self.new_secret = new_secret
        self.reason = reason

__init__(new_secret=None, reason=None)

Initialize the :class:Connection.UpdateSecret class

Source code in pamqp/commands.py

def __init__(self,
             new_secret: str | None = None,
             reason: str | None = None) -> None:
    """Initialize the :class:`Connection.UpdateSecret` class"""
    self.new_secret = new_secret
    self.reason = reason

UpdateSecretOk

Bases: Frame

Update secret response

This method confirms the updated secret is valid.

Source code in pamqp/commands.py

class UpdateSecretOk(base.Frame):
    """Update secret response

    This method confirms the updated secret is valid.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 71  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x000A0047  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Connection.UpdateSecretOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Channel

Work with channels

The channel class provides methods for a client to establish a channel to a server and for both peers to operate the channel thereafter.

Source code in pamqp/commands.py

class Channel:
    """Work with channels

    The channel class provides methods for a client to establish a channel to a
    server and for both peers to operate the channel thereafter.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00140000  # pamqp Mapping Index

    class Open(base.Frame):
        """Open a channel for use

        This method opens a channel to the server.

        :param out_of_band: Protocol level field, do not use, must be ``0``.
            - Default: ``0``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'out_of_band'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0014000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.Open'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Channel.OpenOk']

        # Class Attribute Types for unmarshaling
        _out_of_band = 'shortstr'

        def __init__(self, out_of_band: str = '0') -> None:
            """Initialize the :class:`Channel.Open` class"""
            self.out_of_band = out_of_band
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.out_of_band is not None and self.out_of_band != '0':
                raise ValueError('out_of_band must be 0')

    class OpenOk(base.Frame):
        """Signal that the channel is ready

        This method signals to the client that the channel is ready for use.

        :param channel_id: Deprecated, must be ``0``
            - Default: ``0``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'channel_id'
        ]

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0014000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.OpenOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _channel_id = 'longstr'

        def __init__(self, channel_id: str = '0') -> None:
            """Initialize the :class:`Channel.OpenOk` class"""
            self.channel_id = channel_id
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.channel_id is not None and self.channel_id != '0':
                raise ValueError('channel_id must be 0')

    class Flow(base.Frame):
        """Enable/disable flow from peer

        This method asks the peer to pause or restart the flow of content data
        sent by a consumer. This is a simple flow-control mechanism that a peer
        can use to avoid overflowing its queues or otherwise finding itself
        receiving more messages than it can process. Note that this method is
        not intended for window control. It does not affect contents returned
        by :class:`Basic.GetOk` methods.

        :param active: Start/stop content frames

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'active'
        ]

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00140014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.Flow'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Channel.FlowOk']

        # Class Attribute Types for unmarshaling
        _active = 'bit'

        def __init__(self, active: bool | None = None) -> None:
            """Initialize the :class:`Channel.Flow` class"""
            self.active = active

    class FlowOk(base.Frame):
        """Confirm a flow method

        Confirms to the peer that a flow command was received and processed.

        :param active: Current flow setting

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'active'
        ]

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00140015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.FlowOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _active = 'bit'

        def __init__(self, active: bool | None = None) -> None:
            """Initialize the :class:`Channel.FlowOk` class"""
            self.active = active

    class Close(base.Frame):
        """Request a channel close

        This method indicates that the sender wants to close the channel. This
        may be due to internal conditions (e.g. a forced shut-down) or due to
        an error handling a specific method, i.e. an exception. When a close is
        due to an exception, the sender provides the class and method id of the
        method which caused the exception.

        :param reply_code: Reply code from server
        :param reply_text: Localised reply text
            - Default: ``''``
        :param class_id: Failing method class
        :param method_id: Failing method ID

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'reply_code', 'reply_text', 'class_id', 'method_id'
        ]

        frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00140028  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.Close'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Channel.CloseOk']

        # Class Attribute Types for unmarshaling
        _reply_code = 'short'
        _reply_text = 'shortstr'
        _class_id = 'short'
        _method_id = 'short'

        def __init__(self,
                     reply_code: int | None = None,
                     reply_text: str = '',
                     class_id: int | None = None,
                     method_id: int | None = None) -> None:
            """Initialize the :class:`Channel.Close` class"""
            self.reply_code = reply_code
            self.reply_text = reply_text or ''
            self.class_id = class_id
            self.method_id = method_id

    class CloseOk(base.Frame):
        """Confirm a channel close

        This method confirms a :class:`Channel.Close` method and tells the
        recipient that it is safe to release resources for the channel.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00140029  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Channel.CloseOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Open

Bases: Frame

Open a channel for use

This method opens a channel to the server.

Parameters:

Name	Type	Description	Default
out_of_band	str	Protocol level field, do not use, must be 0. - Default: 0	'0'

Source code in pamqp/commands.py

class Open(base.Frame):
    """Open a channel for use

    This method opens a channel to the server.

    :param out_of_band: Protocol level field, do not use, must be ``0``.
        - Default: ``0``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'out_of_band'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0014000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.Open'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Channel.OpenOk']

    # Class Attribute Types for unmarshaling
    _out_of_band = 'shortstr'

    def __init__(self, out_of_band: str = '0') -> None:
        """Initialize the :class:`Channel.Open` class"""
        self.out_of_band = out_of_band
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.out_of_band is not None and self.out_of_band != '0':
            raise ValueError('out_of_band must be 0')

__init__(out_of_band='0')

Initialize the :class:Channel.Open class

Source code in pamqp/commands.py

def __init__(self, out_of_band: str = '0') -> None:
    """Initialize the :class:`Channel.Open` class"""
    self.out_of_band = out_of_band
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.out_of_band is not None and self.out_of_band != '0':
        raise ValueError('out_of_band must be 0')

OpenOk

Bases: Frame

Signal that the channel is ready

This method signals to the client that the channel is ready for use.

Parameters:

Name	Type	Description	Default
channel_id	str	Deprecated, must be 0 - Default: 0	'0'

Source code in pamqp/commands.py

class OpenOk(base.Frame):
    """Signal that the channel is ready

    This method signals to the client that the channel is ready for use.

    :param channel_id: Deprecated, must be ``0``
        - Default: ``0``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'channel_id'
    ]

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0014000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.OpenOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _channel_id = 'longstr'

    def __init__(self, channel_id: str = '0') -> None:
        """Initialize the :class:`Channel.OpenOk` class"""
        self.channel_id = channel_id
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.channel_id is not None and self.channel_id != '0':
            raise ValueError('channel_id must be 0')

__init__(channel_id='0')

Initialize the :class:Channel.OpenOk class

Source code in pamqp/commands.py

def __init__(self, channel_id: str = '0') -> None:
    """Initialize the :class:`Channel.OpenOk` class"""
    self.channel_id = channel_id
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.channel_id is not None and self.channel_id != '0':
        raise ValueError('channel_id must be 0')

Flow

Bases: Frame

Enable/disable flow from peer

This method asks the peer to pause or restart the flow of content data sent by a consumer. This is a simple flow-control mechanism that a peer can use to avoid overflowing its queues or otherwise finding itself receiving more messages than it can process. Note that this method is not intended for window control. It does not affect contents returned by :class:Basic.GetOk methods.

Parameters:

Name	Type	Description	Default
active	bool | None	Start/stop content frames	None

Source code in pamqp/commands.py

class Flow(base.Frame):
    """Enable/disable flow from peer

    This method asks the peer to pause or restart the flow of content data
    sent by a consumer. This is a simple flow-control mechanism that a peer
    can use to avoid overflowing its queues or otherwise finding itself
    receiving more messages than it can process. Note that this method is
    not intended for window control. It does not affect contents returned
    by :class:`Basic.GetOk` methods.

    :param active: Start/stop content frames

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'active'
    ]

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00140014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.Flow'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Channel.FlowOk']

    # Class Attribute Types for unmarshaling
    _active = 'bit'

    def __init__(self, active: bool | None = None) -> None:
        """Initialize the :class:`Channel.Flow` class"""
        self.active = active

__init__(active=None)

Initialize the :class:Channel.Flow class

Source code in pamqp/commands.py

def __init__(self, active: bool | None = None) -> None:
    """Initialize the :class:`Channel.Flow` class"""
    self.active = active

FlowOk

Bases: Frame

Confirm a flow method

Confirms to the peer that a flow command was received and processed.

Parameters:

Name	Type	Description	Default
active	bool | None	Current flow setting	None

Source code in pamqp/commands.py

class FlowOk(base.Frame):
    """Confirm a flow method

    Confirms to the peer that a flow command was received and processed.

    :param active: Current flow setting

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'active'
    ]

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00140015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.FlowOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _active = 'bit'

    def __init__(self, active: bool | None = None) -> None:
        """Initialize the :class:`Channel.FlowOk` class"""
        self.active = active

__init__(active=None)

Initialize the :class:Channel.FlowOk class

Source code in pamqp/commands.py

def __init__(self, active: bool | None = None) -> None:
    """Initialize the :class:`Channel.FlowOk` class"""
    self.active = active

Close

Bases: Frame

Request a channel close

This method indicates that the sender wants to close the channel. This may be due to internal conditions (e.g. a forced shut-down) or due to an error handling a specific method, i.e. an exception. When a close is due to an exception, the sender provides the class and method id of the method which caused the exception.

Parameters:

Name	Type	Description	Default
reply_code	int | None	Reply code from server	None
reply_text	str	Localised reply text - Default: ''	''
class_id	int | None	Failing method class	None
method_id	int | None	Failing method ID	None

Source code in pamqp/commands.py

class Close(base.Frame):
    """Request a channel close

    This method indicates that the sender wants to close the channel. This
    may be due to internal conditions (e.g. a forced shut-down) or due to
    an error handling a specific method, i.e. an exception. When a close is
    due to an exception, the sender provides the class and method id of the
    method which caused the exception.

    :param reply_code: Reply code from server
    :param reply_text: Localised reply text
        - Default: ``''``
    :param class_id: Failing method class
    :param method_id: Failing method ID

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'reply_code', 'reply_text', 'class_id', 'method_id'
    ]

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00140028  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.Close'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Channel.CloseOk']

    # Class Attribute Types for unmarshaling
    _reply_code = 'short'
    _reply_text = 'shortstr'
    _class_id = 'short'
    _method_id = 'short'

    def __init__(self,
                 reply_code: int | None = None,
                 reply_text: str = '',
                 class_id: int | None = None,
                 method_id: int | None = None) -> None:
        """Initialize the :class:`Channel.Close` class"""
        self.reply_code = reply_code
        self.reply_text = reply_text or ''
        self.class_id = class_id
        self.method_id = method_id

__init__(reply_code=None, reply_text='', class_id=None, method_id=None)

Initialize the :class:Channel.Close class

Source code in pamqp/commands.py

def __init__(self,
             reply_code: int | None = None,
             reply_text: str = '',
             class_id: int | None = None,
             method_id: int | None = None) -> None:
    """Initialize the :class:`Channel.Close` class"""
    self.reply_code = reply_code
    self.reply_text = reply_text or ''
    self.class_id = class_id
    self.method_id = method_id

CloseOk

Bases: Frame

Confirm a channel close

This method confirms a :class:Channel.Close method and tells the recipient that it is safe to release resources for the channel.

Source code in pamqp/commands.py

class CloseOk(base.Frame):
    """Confirm a channel close

    This method confirms a :class:`Channel.Close` method and tells the
    recipient that it is safe to release resources for the channel.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00140029  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Channel.CloseOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Exchange

Work with exchanges

Exchanges match and distribute messages across queues. Exchanges can be configured in the server or declared at runtime.

Source code in pamqp/commands.py

class Exchange:
    """Work with exchanges

    Exchanges match and distribute messages across queues. Exchanges can be
    configured in the server or declared at runtime.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00280000  # pamqp Mapping Index

    class Declare(base.Frame):
        """Verify exchange exists, create if needed

        This method creates an exchange if it does not already exist, and if
        the exchange exists, verifies that it is of the correct and expected
        class.

        .. note:: The AMQP type argument is referred to as "exchange_type" to
            not conflict with the Python type keyword.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param exchange: exchange name
            - Default: ``''``
        :param exchange_type: Exchange type
            - Default: ``direct``
        :param passive: Do not create exchange
            - Default: ``False``
        :param durable: Request a durable exchange
            - Default: ``False``
        :param auto_delete: Auto-delete when unused
            - Default: ``False``
        :param internal: Create internal exchange
            - Default: ``False``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments for declaration
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'exchange', 'exchange_type', 'passive', 'durable',
            'auto_delete', 'internal', 'nowait', 'arguments'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0028000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.Declare'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Exchange.DeclareOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _exchange = 'shortstr'
        _exchange_type = 'shortstr'
        _passive = 'bit'
        _durable = 'bit'
        _auto_delete = 'bit'
        _internal = 'bit'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     exchange: str = '',
                     exchange_type: str = 'direct',
                     passive: bool = False,
                     durable: bool = False,
                     auto_delete: bool = False,
                     internal: bool = False,
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Exchange.Declare` class"""
            self.ticket = ticket
            self.exchange = exchange
            self.exchange_type = exchange_type
            self.passive = passive
            self.durable = durable
            self.auto_delete = auto_delete
            self.internal = internal
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class DeclareOk(base.Frame):
        """Confirm exchange declaration

        This method confirms a Declare method and confirms the name of the
        exchange, essential for automatically-named exchanges.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0028000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.DeclareOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Delete(base.Frame):
        """Delete an exchange

        This method deletes an exchange. When an exchange is deleted all queue
        bindings on the exchange are cancelled.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param exchange: exchange name
            - Default: ``''``
        :param if_unused: Delete only if unused
            - Default: ``False``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'exchange', 'if_unused', 'nowait'
        ]

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00280014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.Delete'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Exchange.DeleteOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _exchange = 'shortstr'
        _if_unused = 'bit'
        _nowait = 'bit'

        def __init__(self,
                     ticket: int = 0,
                     exchange: str = '',
                     if_unused: bool = False,
                     nowait: bool = False) -> None:
            """Initialize the :class:`Exchange.Delete` class"""
            self.ticket = ticket
            self.exchange = exchange
            self.if_unused = if_unused
            self.nowait = nowait
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class DeleteOk(base.Frame):
        """Confirm deletion of an exchange

        This method confirms the deletion of an exchange.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00280015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.DeleteOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Bind(base.Frame):
        """Bind exchange to an exchange

        This method binds an exchange to an exchange.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param destination: Name of the destination exchange to bind to
            - Default: ``''``
        :param source: Name of the source exchange to bind to
            - Default: ``''``
        :param routing_key: Message routing key
            - Default: ``''``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments for binding
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'destination', 'source', 'routing_key', 'nowait',
            'arguments'
        ]

        frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0028001E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.Bind'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Exchange.BindOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _destination = 'shortstr'
        _source = 'shortstr'
        _routing_key = 'shortstr'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     destination: str = '',
                     source: str = '',
                     routing_key: str = '',
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Exchange.Bind` class"""
            self.ticket = ticket
            self.destination = destination
            self.source = source
            self.routing_key = routing_key
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.destination is not None and len(self.destination) > 127:
                raise ValueError('Max length exceeded for destination')
            if self.destination is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.destination):
                raise ValueError('Invalid value for destination')
            if self.source is not None and len(self.source) > 127:
                raise ValueError('Max length exceeded for source')
            if self.source is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.source):
                raise ValueError('Invalid value for source')

    class BindOk(base.Frame):
        """Confirm bind successful

        This method confirms that the bind was successful.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0028001F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.BindOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Unbind(base.Frame):
        """Unbind an exchange from an exchange

        This method unbinds an exchange from an exchange.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param destination: Specifies the name of the destination exchange to
            unbind.
            - Default: ``''``
        :param source: Specifies the name of the source exchange to unbind.
            - Default: ``''``
        :param routing_key: Routing key of binding
            - Default: ``''``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments of binding
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'destination', 'source', 'routing_key', 'nowait',
            'arguments'
        ]

        frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00280028  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.Unbind'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Exchange.UnbindOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _destination = 'shortstr'
        _source = 'shortstr'
        _routing_key = 'shortstr'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     destination: str = '',
                     source: str = '',
                     routing_key: str = '',
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Exchange.Unbind` class"""
            self.ticket = ticket
            self.destination = destination
            self.source = source
            self.routing_key = routing_key
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.destination is not None and len(self.destination) > 127:
                raise ValueError('Max length exceeded for destination')
            if self.destination is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.destination):
                raise ValueError('Invalid value for destination')
            if self.source is not None and len(self.source) > 127:
                raise ValueError('Max length exceeded for source')
            if self.source is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.source):
                raise ValueError('Invalid value for source')

    class UnbindOk(base.Frame):
        """Confirm unbind successful

        This method confirms that the unbind was successful.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00280033  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Exchange.UnbindOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Declare

Bases: Frame

Verify exchange exists, create if needed

This method creates an exchange if it does not already exist, and if the exchange exists, verifies that it is of the correct and expected class.

.. note:: The AMQP type argument is referred to as "exchange_type" to not conflict with the Python type keyword.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
exchange	str	exchange name - Default: ''	''
exchange_type	str	Exchange type - Default: direct	'direct'
passive	bool	Do not create exchange - Default: False	False
durable	bool	Request a durable exchange - Default: False	False
auto_delete	bool	Auto-delete when unused - Default: False	False
internal	bool	Create internal exchange - Default: False	False
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments for declaration - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Declare(base.Frame):
    """Verify exchange exists, create if needed

    This method creates an exchange if it does not already exist, and if
    the exchange exists, verifies that it is of the correct and expected
    class.

    .. note:: The AMQP type argument is referred to as "exchange_type" to
        not conflict with the Python type keyword.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param exchange: exchange name
        - Default: ``''``
    :param exchange_type: Exchange type
        - Default: ``direct``
    :param passive: Do not create exchange
        - Default: ``False``
    :param durable: Request a durable exchange
        - Default: ``False``
    :param auto_delete: Auto-delete when unused
        - Default: ``False``
    :param internal: Create internal exchange
        - Default: ``False``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments for declaration
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'exchange', 'exchange_type', 'passive', 'durable',
        'auto_delete', 'internal', 'nowait', 'arguments'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0028000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.Declare'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Exchange.DeclareOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _exchange = 'shortstr'
    _exchange_type = 'shortstr'
    _passive = 'bit'
    _durable = 'bit'
    _auto_delete = 'bit'
    _internal = 'bit'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 exchange: str = '',
                 exchange_type: str = 'direct',
                 passive: bool = False,
                 durable: bool = False,
                 auto_delete: bool = False,
                 internal: bool = False,
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Exchange.Declare` class"""
        self.ticket = ticket
        self.exchange = exchange
        self.exchange_type = exchange_type
        self.passive = passive
        self.durable = durable
        self.auto_delete = auto_delete
        self.internal = internal
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(ticket=0, exchange='', exchange_type='direct', passive=False, durable=False, auto_delete=False, internal=False, nowait=False, arguments=None)

Initialize the :class:Exchange.Declare class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             exchange: str = '',
             exchange_type: str = 'direct',
             passive: bool = False,
             durable: bool = False,
             auto_delete: bool = False,
             internal: bool = False,
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Exchange.Declare` class"""
    self.ticket = ticket
    self.exchange = exchange
    self.exchange_type = exchange_type
    self.passive = passive
    self.durable = durable
    self.auto_delete = auto_delete
    self.internal = internal
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

DeclareOk

Bases: Frame

Confirm exchange declaration

This method confirms a Declare method and confirms the name of the exchange, essential for automatically-named exchanges.

Source code in pamqp/commands.py

class DeclareOk(base.Frame):
    """Confirm exchange declaration

    This method confirms a Declare method and confirms the name of the
    exchange, essential for automatically-named exchanges.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0028000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.DeclareOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Delete

Bases: Frame

Delete an exchange

This method deletes an exchange. When an exchange is deleted all queue bindings on the exchange are cancelled.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
exchange	str	exchange name - Default: ''	''
if_unused	bool	Delete only if unused - Default: False	False
nowait	bool	Do not send a reply method - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Delete(base.Frame):
    """Delete an exchange

    This method deletes an exchange. When an exchange is deleted all queue
    bindings on the exchange are cancelled.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param exchange: exchange name
        - Default: ``''``
    :param if_unused: Delete only if unused
        - Default: ``False``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'exchange', 'if_unused', 'nowait'
    ]

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00280014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.Delete'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Exchange.DeleteOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _exchange = 'shortstr'
    _if_unused = 'bit'
    _nowait = 'bit'

    def __init__(self,
                 ticket: int = 0,
                 exchange: str = '',
                 if_unused: bool = False,
                 nowait: bool = False) -> None:
        """Initialize the :class:`Exchange.Delete` class"""
        self.ticket = ticket
        self.exchange = exchange
        self.if_unused = if_unused
        self.nowait = nowait
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(ticket=0, exchange='', if_unused=False, nowait=False)

Initialize the :class:Exchange.Delete class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             exchange: str = '',
             if_unused: bool = False,
             nowait: bool = False) -> None:
    """Initialize the :class:`Exchange.Delete` class"""
    self.ticket = ticket
    self.exchange = exchange
    self.if_unused = if_unused
    self.nowait = nowait
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

DeleteOk

Bases: Frame

Confirm deletion of an exchange

This method confirms the deletion of an exchange.

Source code in pamqp/commands.py

class DeleteOk(base.Frame):
    """Confirm deletion of an exchange

    This method confirms the deletion of an exchange.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00280015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.DeleteOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Bind

Bases: Frame

Bind exchange to an exchange

This method binds an exchange to an exchange.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
destination	str	Name of the destination exchange to bind to - Default: ''	''
source	str	Name of the source exchange to bind to - Default: ''	''
routing_key	str	Message routing key - Default: ''	''
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments for binding - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Bind(base.Frame):
    """Bind exchange to an exchange

    This method binds an exchange to an exchange.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param destination: Name of the destination exchange to bind to
        - Default: ``''``
    :param source: Name of the source exchange to bind to
        - Default: ``''``
    :param routing_key: Message routing key
        - Default: ``''``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments for binding
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'destination', 'source', 'routing_key', 'nowait',
        'arguments'
    ]

    frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0028001E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.Bind'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Exchange.BindOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _destination = 'shortstr'
    _source = 'shortstr'
    _routing_key = 'shortstr'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 destination: str = '',
                 source: str = '',
                 routing_key: str = '',
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Exchange.Bind` class"""
        self.ticket = ticket
        self.destination = destination
        self.source = source
        self.routing_key = routing_key
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.destination is not None and len(self.destination) > 127:
            raise ValueError('Max length exceeded for destination')
        if self.destination is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.destination):
            raise ValueError('Invalid value for destination')
        if self.source is not None and len(self.source) > 127:
            raise ValueError('Max length exceeded for source')
        if self.source is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.source):
            raise ValueError('Invalid value for source')

__init__(ticket=0, destination='', source='', routing_key='', nowait=False, arguments=None)

Initialize the :class:Exchange.Bind class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             destination: str = '',
             source: str = '',
             routing_key: str = '',
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Exchange.Bind` class"""
    self.ticket = ticket
    self.destination = destination
    self.source = source
    self.routing_key = routing_key
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.destination is not None and len(self.destination) > 127:
        raise ValueError('Max length exceeded for destination')
    if self.destination is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.destination):
        raise ValueError('Invalid value for destination')
    if self.source is not None and len(self.source) > 127:
        raise ValueError('Max length exceeded for source')
    if self.source is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.source):
        raise ValueError('Invalid value for source')

BindOk

Bases: Frame

Confirm bind successful

This method confirms that the bind was successful.

Source code in pamqp/commands.py

class BindOk(base.Frame):
    """Confirm bind successful

    This method confirms that the bind was successful.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0028001F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.BindOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Unbind

Bases: Frame

Unbind an exchange from an exchange

This method unbinds an exchange from an exchange.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
destination	str	Specifies the name of the destination exchange to unbind. - Default: ''	''
source	str	Specifies the name of the source exchange to unbind. - Default: ''	''
routing_key	str	Routing key of binding - Default: ''	''
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments of binding - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Unbind(base.Frame):
    """Unbind an exchange from an exchange

    This method unbinds an exchange from an exchange.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param destination: Specifies the name of the destination exchange to
        unbind.
        - Default: ``''``
    :param source: Specifies the name of the source exchange to unbind.
        - Default: ``''``
    :param routing_key: Routing key of binding
        - Default: ``''``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments of binding
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'destination', 'source', 'routing_key', 'nowait',
        'arguments'
    ]

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00280028  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.Unbind'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Exchange.UnbindOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _destination = 'shortstr'
    _source = 'shortstr'
    _routing_key = 'shortstr'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 destination: str = '',
                 source: str = '',
                 routing_key: str = '',
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Exchange.Unbind` class"""
        self.ticket = ticket
        self.destination = destination
        self.source = source
        self.routing_key = routing_key
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.destination is not None and len(self.destination) > 127:
            raise ValueError('Max length exceeded for destination')
        if self.destination is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.destination):
            raise ValueError('Invalid value for destination')
        if self.source is not None and len(self.source) > 127:
            raise ValueError('Max length exceeded for source')
        if self.source is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.source):
            raise ValueError('Invalid value for source')

__init__(ticket=0, destination='', source='', routing_key='', nowait=False, arguments=None)

Initialize the :class:Exchange.Unbind class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             destination: str = '',
             source: str = '',
             routing_key: str = '',
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Exchange.Unbind` class"""
    self.ticket = ticket
    self.destination = destination
    self.source = source
    self.routing_key = routing_key
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.destination is not None and len(self.destination) > 127:
        raise ValueError('Max length exceeded for destination')
    if self.destination is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.destination):
        raise ValueError('Invalid value for destination')
    if self.source is not None and len(self.source) > 127:
        raise ValueError('Max length exceeded for source')
    if self.source is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.source):
        raise ValueError('Invalid value for source')

UnbindOk

Bases: Frame

Confirm unbind successful

This method confirms that the unbind was successful.

Source code in pamqp/commands.py

class UnbindOk(base.Frame):
    """Confirm unbind successful

    This method confirms that the unbind was successful.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00280033  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Exchange.UnbindOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Queue

Work with queues

Queues store and forward messages. Queues can be configured in the server or created at runtime. Queues must be attached to at least one exchange in order to receive messages from publishers.

Source code in pamqp/commands.py

class Queue:
    """Work with queues

    Queues store and forward messages. Queues can be configured in the server
    or created at runtime. Queues must be attached to at least one exchange in
    order to receive messages from publishers.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320000  # pamqp Mapping Index

    class Declare(base.Frame):
        """Declare queue, create if needed

        This method creates or checks a queue. When creating a new queue the
        client can specify various properties that control the durability of
        the queue and its contents, and the level of sharing for the queue.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: queue name
            - Default: ``''``
        :param passive: Do not create queue
            - Default: ``False``
        :param durable: Request a durable queue
            - Default: ``False``
        :param exclusive: Request an exclusive queue
            - Default: ``False``
        :param auto_delete: Auto-delete queue when unused
            - Default: ``False``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments for declaration
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'passive', 'durable', 'exclusive',
            'auto_delete', 'nowait', 'arguments'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0032000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.Declare'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Queue.DeclareOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _passive = 'bit'
        _durable = 'bit'
        _exclusive = 'bit'
        _auto_delete = 'bit'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     passive: bool = False,
                     durable: bool = False,
                     exclusive: bool = False,
                     auto_delete: bool = False,
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Queue.Declare` class"""
            self.ticket = ticket
            self.queue = queue
            self.passive = passive
            self.durable = durable
            self.exclusive = exclusive
            self.auto_delete = auto_delete
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class DeclareOk(base.Frame):
        """Confirms a queue definition

        This method confirms a Declare method and confirms the name of the
        queue, essential for automatically-named queues.

        :param queue: Reports the name of the queue. If the server generated a
            queue name, this field contains that name.
        :param message_count: Number of messages in queue
        :param consumer_count: Number of consumers
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'queue', 'message_count', 'consumer_count'
        ]

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0032000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.DeclareOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _queue = 'shortstr'
        _message_count = 'long'
        _consumer_count = 'long'

        def __init__(self,
                     queue: str | None = None,
                     message_count: int | None = None,
                     consumer_count: int | None = None) -> None:
            """Initialize the :class:`Queue.DeclareOk` class"""
            self.queue = queue
            self.message_count = message_count
            self.consumer_count = consumer_count
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class Bind(base.Frame):
        """Bind queue to an exchange

        This method binds a queue to an exchange. Until a queue is bound it
        will not receive any messages. In a classic messaging model, store-and-
        forward queues are bound to a direct exchange and subscription queues
        are bound to a topic exchange.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to bind.
            - Default: ``''``
        :param exchange: Name of the exchange to bind to
            - Default: ``''``
        :param routing_key: Message routing key
            - Default: ``''``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments for binding
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'exchange', 'routing_key', 'nowait', 'arguments'
        ]

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.Bind'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Queue.BindOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     exchange: str = '',
                     routing_key: str = '',
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Queue.Bind` class"""
            self.ticket = ticket
            self.queue = queue
            self.exchange = exchange
            self.routing_key = routing_key
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class BindOk(base.Frame):
        """Confirm bind successful

        This method confirms that the bind was successful.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.BindOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Purge(base.Frame):
        """Purge a queue

        This method removes all messages from a queue which are not awaiting
        acknowledgment.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to purge.
            - Default: ``''``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'nowait'
        ]

        frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0032001E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.Purge'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Queue.PurgeOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _nowait = 'bit'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     nowait: bool = False) -> None:
            """Initialize the :class:`Queue.Purge` class"""
            self.ticket = ticket
            self.queue = queue
            self.nowait = nowait
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class PurgeOk(base.Frame):
        """Confirms a queue purge

        This method confirms the purge of a queue.

        :param message_count: Reports the number of messages purged.

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'message_count'
        ]

        frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0032001F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.PurgeOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _message_count = 'long'

        def __init__(self, message_count: int | None = None) -> None:
            """Initialize the :class:`Queue.PurgeOk` class"""
            self.message_count = message_count

    class Delete(base.Frame):
        """Delete a queue

        This method deletes a queue. When a queue is deleted any pending
        messages are sent to a dead-letter queue if this is defined in the
        server configuration, and all consumers on the queue are cancelled.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to delete.
            - Default: ``''``
        :param if_unused: Delete only if unused
            - Default: ``False``
        :param if_empty: Delete only if empty
            - Default: ``False``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'if_unused', 'if_empty', 'nowait'
        ]

        frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320028  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.Delete'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Queue.DeleteOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _if_unused = 'bit'
        _if_empty = 'bit'
        _nowait = 'bit'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     if_unused: bool = False,
                     if_empty: bool = False,
                     nowait: bool = False) -> None:
            """Initialize the :class:`Queue.Delete` class"""
            self.ticket = ticket
            self.queue = queue
            self.if_unused = if_unused
            self.if_empty = if_empty
            self.nowait = nowait
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class DeleteOk(base.Frame):
        """Confirm deletion of a queue

        This method confirms the deletion of a queue.

        :param message_count: Reports the number of messages deleted.

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'message_count'
        ]

        frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320029  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.DeleteOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _message_count = 'long'

        def __init__(self, message_count: int | None = None) -> None:
            """Initialize the :class:`Queue.DeleteOk` class"""
            self.message_count = message_count

    class Unbind(base.Frame):
        """Unbind a queue from an exchange

        This method unbinds a queue from an exchange.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to unbind.
            - Default: ``''``
        :param exchange: The name of the exchange to unbind from.
            - Default: ``''``
        :param routing_key: Routing key of binding
            - Default: ``''``
        :param arguments: Arguments of binding
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'exchange', 'routing_key', 'arguments'
        ]

        frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320032  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.Unbind'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Queue.UnbindOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     exchange: str = '',
                     routing_key: str = '',
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Queue.Unbind` class"""
            self.ticket = ticket
            self.queue = queue
            self.exchange = exchange
            self.routing_key = routing_key
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class UnbindOk(base.Frame):
        """Confirm unbind successful

        This method confirms that the unbind was successful.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x00320033  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Queue.UnbindOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Declare

Bases: Frame

Declare queue, create if needed

This method creates or checks a queue. When creating a new queue the client can specify various properties that control the durability of the queue and its contents, and the level of sharing for the queue.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	queue name - Default: ''	''
passive	bool	Do not create queue - Default: False	False
durable	bool	Request a durable queue - Default: False	False
exclusive	bool	Request an exclusive queue - Default: False	False
auto_delete	bool	Auto-delete queue when unused - Default: False	False
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments for declaration - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Declare(base.Frame):
    """Declare queue, create if needed

    This method creates or checks a queue. When creating a new queue the
    client can specify various properties that control the durability of
    the queue and its contents, and the level of sharing for the queue.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: queue name
        - Default: ``''``
    :param passive: Do not create queue
        - Default: ``False``
    :param durable: Request a durable queue
        - Default: ``False``
    :param exclusive: Request an exclusive queue
        - Default: ``False``
    :param auto_delete: Auto-delete queue when unused
        - Default: ``False``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments for declaration
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'passive', 'durable', 'exclusive',
        'auto_delete', 'nowait', 'arguments'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0032000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.Declare'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Queue.DeclareOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _passive = 'bit'
    _durable = 'bit'
    _exclusive = 'bit'
    _auto_delete = 'bit'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 passive: bool = False,
                 durable: bool = False,
                 exclusive: bool = False,
                 auto_delete: bool = False,
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Queue.Declare` class"""
        self.ticket = ticket
        self.queue = queue
        self.passive = passive
        self.durable = durable
        self.exclusive = exclusive
        self.auto_delete = auto_delete
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(ticket=0, queue='', passive=False, durable=False, exclusive=False, auto_delete=False, nowait=False, arguments=None)

Initialize the :class:Queue.Declare class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             passive: bool = False,
             durable: bool = False,
             exclusive: bool = False,
             auto_delete: bool = False,
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Queue.Declare` class"""
    self.ticket = ticket
    self.queue = queue
    self.passive = passive
    self.durable = durable
    self.exclusive = exclusive
    self.auto_delete = auto_delete
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

DeclareOk

Bases: Frame

Confirms a queue definition

This method confirms a Declare method and confirms the name of the queue, essential for automatically-named queues.

Parameters:

Name	Type	Description	Default
queue	str | None	Reports the name of the queue. If the server generated a queue name, this field contains that name.	None
message_count	int | None	Number of messages in queue	None
consumer_count	int | None	Number of consumers	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class DeclareOk(base.Frame):
    """Confirms a queue definition

    This method confirms a Declare method and confirms the name of the
    queue, essential for automatically-named queues.

    :param queue: Reports the name of the queue. If the server generated a
        queue name, this field contains that name.
    :param message_count: Number of messages in queue
    :param consumer_count: Number of consumers
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'queue', 'message_count', 'consumer_count'
    ]

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0032000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.DeclareOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _queue = 'shortstr'
    _message_count = 'long'
    _consumer_count = 'long'

    def __init__(self,
                 queue: str | None = None,
                 message_count: int | None = None,
                 consumer_count: int | None = None) -> None:
        """Initialize the :class:`Queue.DeclareOk` class"""
        self.queue = queue
        self.message_count = message_count
        self.consumer_count = consumer_count
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(queue=None, message_count=None, consumer_count=None)

Initialize the :class:Queue.DeclareOk class

Source code in pamqp/commands.py

def __init__(self,
             queue: str | None = None,
             message_count: int | None = None,
             consumer_count: int | None = None) -> None:
    """Initialize the :class:`Queue.DeclareOk` class"""
    self.queue = queue
    self.message_count = message_count
    self.consumer_count = consumer_count
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

Bind

Bases: Frame

Bind queue to an exchange

This method binds a queue to an exchange. Until a queue is bound it will not receive any messages. In a classic messaging model, store-and- forward queues are bound to a direct exchange and subscription queues are bound to a topic exchange.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to bind. - Default: ''	''
exchange	str	Name of the exchange to bind to - Default: ''	''
routing_key	str	Message routing key - Default: ''	''
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments for binding - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Bind(base.Frame):
    """Bind queue to an exchange

    This method binds a queue to an exchange. Until a queue is bound it
    will not receive any messages. In a classic messaging model, store-and-
    forward queues are bound to a direct exchange and subscription queues
    are bound to a topic exchange.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to bind.
        - Default: ``''``
    :param exchange: Name of the exchange to bind to
        - Default: ``''``
    :param routing_key: Message routing key
        - Default: ``''``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments for binding
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'exchange', 'routing_key', 'nowait', 'arguments'
    ]

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.Bind'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Queue.BindOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 exchange: str = '',
                 routing_key: str = '',
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Queue.Bind` class"""
        self.ticket = ticket
        self.queue = queue
        self.exchange = exchange
        self.routing_key = routing_key
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(ticket=0, queue='', exchange='', routing_key='', nowait=False, arguments=None)

Initialize the :class:Queue.Bind class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             exchange: str = '',
             routing_key: str = '',
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Queue.Bind` class"""
    self.ticket = ticket
    self.queue = queue
    self.exchange = exchange
    self.routing_key = routing_key
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

BindOk

Bases: Frame

Confirm bind successful

This method confirms that the bind was successful.

Source code in pamqp/commands.py

class BindOk(base.Frame):
    """Confirm bind successful

    This method confirms that the bind was successful.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.BindOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Purge

Bases: Frame

Purge a queue

This method removes all messages from a queue which are not awaiting acknowledgment.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to purge. - Default: ''	''
nowait	bool	Do not send a reply method - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Purge(base.Frame):
    """Purge a queue

    This method removes all messages from a queue which are not awaiting
    acknowledgment.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to purge.
        - Default: ``''``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'nowait'
    ]

    frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0032001E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.Purge'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Queue.PurgeOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _nowait = 'bit'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 nowait: bool = False) -> None:
        """Initialize the :class:`Queue.Purge` class"""
        self.ticket = ticket
        self.queue = queue
        self.nowait = nowait
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(ticket=0, queue='', nowait=False)

Initialize the :class:Queue.Purge class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             nowait: bool = False) -> None:
    """Initialize the :class:`Queue.Purge` class"""
    self.ticket = ticket
    self.queue = queue
    self.nowait = nowait
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

PurgeOk

Bases: Frame

Confirms a queue purge

This method confirms the purge of a queue.

Parameters:

Name	Type	Description	Default
message_count	int | None	Reports the number of messages purged.	None

Source code in pamqp/commands.py

class PurgeOk(base.Frame):
    """Confirms a queue purge

    This method confirms the purge of a queue.

    :param message_count: Reports the number of messages purged.

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'message_count'
    ]

    frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0032001F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.PurgeOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _message_count = 'long'

    def __init__(self, message_count: int | None = None) -> None:
        """Initialize the :class:`Queue.PurgeOk` class"""
        self.message_count = message_count

__init__(message_count=None)

Initialize the :class:Queue.PurgeOk class

Source code in pamqp/commands.py

def __init__(self, message_count: int | None = None) -> None:
    """Initialize the :class:`Queue.PurgeOk` class"""
    self.message_count = message_count

Delete

Bases: Frame

Delete a queue

This method deletes a queue. When a queue is deleted any pending messages are sent to a dead-letter queue if this is defined in the server configuration, and all consumers on the queue are cancelled.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to delete. - Default: ''	''
if_unused	bool	Delete only if unused - Default: False	False
if_empty	bool	Delete only if empty - Default: False	False
nowait	bool	Do not send a reply method - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Delete(base.Frame):
    """Delete a queue

    This method deletes a queue. When a queue is deleted any pending
    messages are sent to a dead-letter queue if this is defined in the
    server configuration, and all consumers on the queue are cancelled.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to delete.
        - Default: ``''``
    :param if_unused: Delete only if unused
        - Default: ``False``
    :param if_empty: Delete only if empty
        - Default: ``False``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'if_unused', 'if_empty', 'nowait'
    ]

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320028  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.Delete'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Queue.DeleteOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _if_unused = 'bit'
    _if_empty = 'bit'
    _nowait = 'bit'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 if_unused: bool = False,
                 if_empty: bool = False,
                 nowait: bool = False) -> None:
        """Initialize the :class:`Queue.Delete` class"""
        self.ticket = ticket
        self.queue = queue
        self.if_unused = if_unused
        self.if_empty = if_empty
        self.nowait = nowait
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(ticket=0, queue='', if_unused=False, if_empty=False, nowait=False)

Initialize the :class:Queue.Delete class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             if_unused: bool = False,
             if_empty: bool = False,
             nowait: bool = False) -> None:
    """Initialize the :class:`Queue.Delete` class"""
    self.ticket = ticket
    self.queue = queue
    self.if_unused = if_unused
    self.if_empty = if_empty
    self.nowait = nowait
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

DeleteOk

Bases: Frame

Confirm deletion of a queue

This method confirms the deletion of a queue.

Parameters:

Name	Type	Description	Default
message_count	int | None	Reports the number of messages deleted.	None

Source code in pamqp/commands.py

class DeleteOk(base.Frame):
    """Confirm deletion of a queue

    This method confirms the deletion of a queue.

    :param message_count: Reports the number of messages deleted.

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'message_count'
    ]

    frame_id: typing.ClassVar[int] = 41  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320029  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.DeleteOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _message_count = 'long'

    def __init__(self, message_count: int | None = None) -> None:
        """Initialize the :class:`Queue.DeleteOk` class"""
        self.message_count = message_count

__init__(message_count=None)

Initialize the :class:Queue.DeleteOk class

Source code in pamqp/commands.py

def __init__(self, message_count: int | None = None) -> None:
    """Initialize the :class:`Queue.DeleteOk` class"""
    self.message_count = message_count

Unbind

Bases: Frame

Unbind a queue from an exchange

This method unbinds a queue from an exchange.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to unbind. - Default: ''	''
exchange	str	The name of the exchange to unbind from. - Default: ''	''
routing_key	str	Routing key of binding - Default: ''	''
arguments	Arguments | None	Arguments of binding - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Unbind(base.Frame):
    """Unbind a queue from an exchange

    This method unbinds a queue from an exchange.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to unbind.
        - Default: ``''``
    :param exchange: The name of the exchange to unbind from.
        - Default: ``''``
    :param routing_key: Routing key of binding
        - Default: ``''``
    :param arguments: Arguments of binding
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'exchange', 'routing_key', 'arguments'
    ]

    frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320032  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.Unbind'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Queue.UnbindOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 exchange: str = '',
                 routing_key: str = '',
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Queue.Unbind` class"""
        self.ticket = ticket
        self.queue = queue
        self.exchange = exchange
        self.routing_key = routing_key
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(ticket=0, queue='', exchange='', routing_key='', arguments=None)

Initialize the :class:Queue.Unbind class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             exchange: str = '',
             routing_key: str = '',
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Queue.Unbind` class"""
    self.ticket = ticket
    self.queue = queue
    self.exchange = exchange
    self.routing_key = routing_key
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

UnbindOk

Bases: Frame

Confirm unbind successful

This method confirms that the unbind was successful.

Source code in pamqp/commands.py

class UnbindOk(base.Frame):
    """Confirm unbind successful

    This method confirms that the unbind was successful.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 51  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00320033  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Queue.UnbindOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Basic

Work with basic content

The Basic class provides methods that support an industry-standard messaging model.

Source code in pamqp/commands.py

class Basic:
    """Work with basic content

    The Basic class provides methods that support an industry-standard
    messaging model.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0000  # pamqp Mapping Index

    class Qos(base.Frame):
        """Specify quality of service

        This method requests a specific quality of service. The QoS can be
        specified for the current channel or for all channels on the
        connection. The particular properties and semantics of a qos method
        always depend on the content class semantics. Though the qos method
        could in principle apply to both peers, it is currently meaningful only
        for the server.

        :param prefetch_size: Prefetch window in octets
            - Default: ``0``
        :param prefetch_count: Prefetch window in messages
            - Default: ``0``
        :param global_: Apply to entire connection
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'prefetch_size', 'prefetch_count', 'global_'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Qos'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Basic.QosOk']

        # Class Attribute Types for unmarshaling
        _prefetch_size = 'long'
        _prefetch_count = 'short'
        _global_ = 'bit'

        def __init__(self,
                     prefetch_size: int = 0,
                     prefetch_count: int = 0,
                     global_: bool = False) -> None:
            """Initialize the :class:`Basic.Qos` class"""
            self.prefetch_size = prefetch_size
            self.prefetch_count = prefetch_count
            self.global_ = global_

    class QosOk(base.Frame):
        """Confirm the requested qos

        This method tells the client that the requested QoS levels could be
        handled by the server. The requested QoS applies to all active
        consumers until a new QoS is defined.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.QosOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Consume(base.Frame):
        """Start a queue consumer

        This method asks the server to start a "consumer", which is a transient
        request for messages from a specific queue. Consumers last as long as
        the channel they were declared on, or until the client cancels them.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to consume from.
            - Default: ``''``
        :param consumer_tag: Specifies the identifier for the consumer. The
            consumer tag is local to a channel, so two clients can use the same
            consumer tags. If this field is empty the server will generate a
            unique tag.
            - Default: ``''``
        :param no_local: Do not deliver own messages
            - Default: ``False``
        :param no_ack: No acknowledgement needed
            - Default: ``False``
        :param exclusive: Request exclusive access
            - Default: ``False``
        :param nowait: Do not send a reply method
            - Default: ``False``
        :param arguments: Arguments for declaration
            - Default: ``{}``
        :type arguments: :const:`~pamqp.common.Arguments`
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'consumer_tag', 'no_local', 'no_ack',
            'exclusive', 'nowait', 'arguments'
        ]

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Consume'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Basic.ConsumeOk']

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _consumer_tag = 'shortstr'
        _no_local = 'bit'
        _no_ack = 'bit'
        _exclusive = 'bit'
        _nowait = 'bit'
        _arguments = 'table'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     consumer_tag: str = '',
                     no_local: bool = False,
                     no_ack: bool = False,
                     exclusive: bool = False,
                     nowait: bool = False,
                     arguments: common.Arguments | None = None) -> None:
            """Initialize the :class:`Basic.Consume` class"""
            self.ticket = ticket
            self.queue = queue
            self.consumer_tag = consumer_tag
            self.no_local = no_local
            self.no_ack = no_ack
            self.exclusive = exclusive
            self.nowait = nowait
            self.arguments = arguments or {}
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class ConsumeOk(base.Frame):
        """Confirm a new consumer

        The server provides the client with a consumer tag, which is used by
        the client for methods called on the consumer at a later stage.

        :param consumer_tag: Holds the consumer tag specified by the client or
            provided by the server.

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'consumer_tag'
        ]

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.ConsumeOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _consumer_tag = 'shortstr'

        def __init__(self, consumer_tag: str | None = None) -> None:
            """Initialize the :class:`Basic.ConsumeOk` class"""
            self.consumer_tag = consumer_tag

    class Cancel(base.Frame):
        """End a queue consumer

        This method cancels a consumer. This does not affect already delivered
        messages, but it does mean the server will not send any more messages
        for that consumer. The client may receive an arbitrary number of
        messages in between sending the cancel method and receiving the cancel-
        ok reply.  It may also be sent from the server to the client in the
        event of the consumer being unexpectedly cancelled (i.e. cancelled for
        any reason other than the server receiving the corresponding
        basic.cancel from the client). This allows clients to be notified of
        the loss of consumers due to events such as queue deletion. Note that
        as it is not a MUST for clients to accept this method from the server,
        it is advisable for the broker to be able to identify those clients
        that are capable of accepting the method, through some means of
        capability negotiation.

        :param consumer_tag: Consumer tag
        :param nowait: Do not send a reply method
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'consumer_tag', 'nowait'
        ]

        frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C001E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Cancel'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Basic.CancelOk']

        # Class Attribute Types for unmarshaling
        _consumer_tag = 'shortstr'
        _nowait = 'bit'

        def __init__(self,
                     consumer_tag: str | None = None,
                     nowait: bool = False) -> None:
            """Initialize the :class:`Basic.Cancel` class"""
            self.consumer_tag = consumer_tag
            self.nowait = nowait if nowait is not None else False

    class CancelOk(base.Frame):
        """Confirm a cancelled consumer

        This method confirms that the cancellation was completed.

        :param consumer_tag: Consumer tag

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'consumer_tag'
        ]

        frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C001F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.CancelOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _consumer_tag = 'shortstr'

        def __init__(self, consumer_tag: str | None = None) -> None:
            """Initialize the :class:`Basic.CancelOk` class"""
            self.consumer_tag = consumer_tag

    class Publish(base.Frame):
        """Publish a message

        This method publishes a message to a specific exchange. The message
        will be routed to queues as defined by the exchange configuration and
        distributed to any active consumers when the transaction, if any, is
        committed.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param exchange: Specifies the name of the exchange to publish to. The
            exchange name can be empty, meaning the default exchange. If the
            exchange name is specified, and that exchange does not exist, the
            server will raise a channel exception.
            - Default: ``''``
        :param routing_key: Message routing key
            - Default: ``''``
        :param mandatory: Indicate mandatory routing
            - Default: ``False``
        :param immediate: Request immediate delivery
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'exchange', 'routing_key', 'mandatory', 'immediate'
        ]

        frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0028  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Publish'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'
        _mandatory = 'bit'
        _immediate = 'bit'

        def __init__(self,
                     ticket: int = 0,
                     exchange: str = '',
                     routing_key: str = '',
                     mandatory: bool = False,
                     immediate: bool = False) -> None:
            """Initialize the :class:`Basic.Publish` class"""
            self.ticket = ticket
            self.exchange = exchange
            self.routing_key = routing_key
            self.mandatory = mandatory
            self.immediate = immediate
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class Return(base.Frame):
        """Return a failed message

        This method returns an undeliverable message that was published with
        the "immediate" flag set, or an unroutable message published with the
        "mandatory" flag set. The reply code and text provide information about
        the reason that the message was undeliverable.

        :param reply_code: Reply code from server
        :param reply_text: Localised reply text
            - Default: ``''``
        :param exchange: Specifies the name of the exchange that the message
            was originally published to.  May be empty, meaning the default
            exchange.
            - Default: ``''``
        :param routing_key: Message routing key
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'reply_code', 'reply_text', 'exchange', 'routing_key'
        ]

        frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0032  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Return'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _reply_code = 'short'
        _reply_text = 'shortstr'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'

        def __init__(self,
                     reply_code: int | None = None,
                     reply_text: str = '',
                     exchange: str = '',
                     routing_key: str | None = None) -> None:
            """Initialize the :class:`Basic.Return` class"""
            self.reply_code = reply_code
            self.reply_text = reply_text or ''
            self.exchange = exchange or ''
            self.routing_key = routing_key
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class Deliver(base.Frame):
        """Notify the client of a consumer message

        This method delivers a message to the client, via a consumer. In the
        asynchronous message delivery model, the client starts a consumer using
        the Consume method, then the server responds with Deliver methods as
        and when messages arrive for that consumer.

        :param consumer_tag: Consumer tag
        :param delivery_tag: Server-assigned delivery tag
        :param redelivered: Message is being redelivered
            - Default: ``False``
        :param exchange: Specifies the name of the exchange that the message
            was originally published to. May be empty, indicating the default
            exchange.
            - Default: ``''``
        :param routing_key: Message routing key
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'consumer_tag', 'delivery_tag', 'redelivered', 'exchange',
            'routing_key'
        ]

        frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C003C  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Deliver'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _consumer_tag = 'shortstr'
        _delivery_tag = 'longlong'
        _redelivered = 'bit'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'

        def __init__(self,
                     consumer_tag: str | None = None,
                     delivery_tag: int | None = None,
                     redelivered: bool = False,
                     exchange: str = '',
                     routing_key: str | None = None) -> None:
            """Initialize the :class:`Basic.Deliver` class"""
            self.consumer_tag = consumer_tag
            self.delivery_tag = delivery_tag
            self.redelivered = redelivered if redelivered is not None else False
            self.exchange = exchange or ''
            self.routing_key = routing_key
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class Get(base.Frame):
        """Direct access to a queue

        This method provides a direct access to the messages in a queue using a
        synchronous dialogue that is designed for specific types of application
        where synchronous functionality is more important than performance.

        :param ticket: Deprecated, must be ``0``
            - Default: ``0``
        :param queue: Specifies the name of the queue to get a message from.
            - Default: ``''``
        :param no_ack: No acknowledgement needed
            - Default: ``False``
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'ticket', 'queue', 'no_ack'
        ]

        frame_id: typing.ClassVar[int] = 70  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0046  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Get'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = [
            'Basic.GetOk', 'Basic.GetEmpty'
        ]

        # Class Attribute Types for unmarshaling
        _ticket = 'short'
        _queue = 'shortstr'
        _no_ack = 'bit'

        def __init__(self,
                     ticket: int = 0,
                     queue: str = '',
                     no_ack: bool = False) -> None:
            """Initialize the :class:`Basic.Get` class"""
            self.ticket = ticket
            self.queue = queue
            self.no_ack = no_ack
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.ticket is not None and self.ticket != 0:
                raise ValueError('ticket must be 0')
            if self.queue is not None and len(self.queue) > 256:
                raise ValueError('Max length exceeded for queue')
            if self.queue is not None and not constants.DOMAIN_REGEX[
                    'queue-name'].fullmatch(self.queue):
                raise ValueError('Invalid value for queue')

    class GetOk(base.Frame):
        """Provide client with a message

        This method delivers a message to the client following a get method. A
        message delivered by 'get-ok' must be acknowledged unless the no-ack
        option was set in the get method.

        :param delivery_tag: Server-assigned delivery tag
        :param redelivered: Message is being redelivered
            - Default: ``False``
        :param exchange: Specifies the name of the exchange that the message
            was originally published to. If empty, the message was published to
            the default exchange.
            - Default: ``''``
        :param routing_key: Message routing key
        :param message_count: Number of messages in queue
        :raises ValueError: when an argument fails to validate

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'delivery_tag', 'redelivered', 'exchange', 'routing_key',
            'message_count'
        ]

        frame_id: typing.ClassVar[int] = 71  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0047  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.GetOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _delivery_tag = 'longlong'
        _redelivered = 'bit'
        _exchange = 'shortstr'
        _routing_key = 'shortstr'
        _message_count = 'long'

        def __init__(self,
                     delivery_tag: int | None = None,
                     redelivered: bool = False,
                     exchange: str = '',
                     routing_key: str | None = None,
                     message_count: int | None = None) -> None:
            """Initialize the :class:`Basic.GetOk` class"""
            self.delivery_tag = delivery_tag
            self.redelivered = redelivered if redelivered is not None else False
            self.exchange = exchange or ''
            self.routing_key = routing_key
            self.message_count = message_count
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.exchange is not None and len(self.exchange) > 127:
                raise ValueError('Max length exceeded for exchange')
            if self.exchange is not None and not constants.DOMAIN_REGEX[
                    'exchange-name'].fullmatch(self.exchange):
                raise ValueError('Invalid value for exchange')

    class GetEmpty(base.Frame):
        """Indicate no messages available

        This method tells the client that the queue has no messages available
        for the client.

        :param cluster_id: Deprecated, must be empty
            - Default: ``''``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'cluster_id'
        ]

        frame_id: typing.ClassVar[int] = 72  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0048  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.GetEmpty'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _cluster_id = 'shortstr'

        def __init__(self, cluster_id: str = '') -> None:
            """Initialize the :class:`Basic.GetEmpty` class"""
            self.cluster_id = cluster_id
            self.validate()

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

            :raises ValueError: on validation error

            """
            if self.cluster_id is not None and self.cluster_id != '':
                raise ValueError('cluster_id must be empty')

    class Ack(base.Frame):
        """Acknowledge one or more messages

        When sent by the client, this method acknowledges one or more messages
        delivered via the Deliver or Get-Ok methods.  When sent by server, this
        method acknowledges one or more messages published with the Publish
        method on a channel in confirm mode.  The acknowledgement can be for a
        single message or a set of messages up to and including a specific
        message.

        :param delivery_tag: Server-assigned delivery tag
            - Default: ``0``
        :param multiple: Acknowledge multiple messages
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'delivery_tag', 'multiple'
        ]

        frame_id: typing.ClassVar[int] = 80  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0050  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Ack'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _delivery_tag = 'longlong'
        _multiple = 'bit'

        def __init__(self,
                     delivery_tag: int = 0,
                     multiple: bool = False) -> None:
            """Initialize the :class:`Basic.Ack` class"""
            self.delivery_tag = delivery_tag
            self.multiple = multiple

    class Reject(base.Frame):
        """Reject an incoming message

        This method allows a client to reject a message. It can be used to
        interrupt and cancel large incoming messages, or return untreatable
        messages to their original queue.

        :param delivery_tag: Server-assigned delivery tag
        :param requeue: Requeue the message
            - Default: ``True``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'delivery_tag', 'requeue'
        ]

        frame_id: typing.ClassVar[int] = 90  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C005A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Reject'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _delivery_tag = 'longlong'
        _requeue = 'bit'

        def __init__(self,
                     delivery_tag: int | None = None,
                     requeue: bool = True) -> None:
            """Initialize the :class:`Basic.Reject` class"""
            self.delivery_tag = delivery_tag
            self.requeue = requeue if requeue is not None else True

    class RecoverAsync(base.Frame):
        """Redeliver unacknowledged messages

        This method asks the server to redeliver all unacknowledged messages on
        a specified channel. Zero or more messages may be redelivered.  This
        method is deprecated in favour of the synchronous Recover/Recover-Ok.

        .. deprecated:: This command is deprecated in AMQP 0-9-1

        :param requeue: Requeue the message
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'requeue'
        ]

        frame_id: typing.ClassVar[int] = 100  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0064  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.RecoverAsync'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _requeue = 'bit'

        def __init__(self, requeue: bool = False) -> None:
            """Initialize the :class:`Basic.RecoverAsync` class"""
            self.requeue = requeue
            warnings.warn(constants.DEPRECATION_WARNING,
                          category=DeprecationWarning)

    class Recover(base.Frame):
        """Redeliver unacknowledged messages

        This method asks the server to redeliver all unacknowledged messages on
        a specified channel. Zero or more messages may be redelivered.  This
        method replaces the asynchronous Recover.

        :param requeue: Requeue the message
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'requeue'
        ]

        frame_id: typing.ClassVar[int] = 110  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C006E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Recover'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Basic.RecoverOk']

        # Class Attribute Types for unmarshaling
        _requeue = 'bit'

        def __init__(self, requeue: bool = False) -> None:
            """Initialize the :class:`Basic.Recover` class"""
            self.requeue = requeue

    class RecoverOk(base.Frame):
        """Confirm recovery

        This method acknowledges a :class:`Basic.Recover` method.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 111  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C006F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.RecoverOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Nack(base.Frame):
        """Reject one or more incoming messages

        This method allows a client to reject one or more incoming messages. It
        can be used to interrupt and cancel large incoming messages, or return
        untreatable messages to their original queue.  This method is also used
        by the server to inform publishers on channels in confirm mode of
        unhandled messages.  If a publisher receives this method, it probably
        needs to republish the offending messages.

        :param delivery_tag: Server-assigned delivery tag
            - Default: ``0``
        :param multiple: Reject multiple messages
            - Default: ``False``
        :param requeue: Requeue the message
            - Default: ``True``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'delivery_tag', 'multiple', 'requeue'
        ]

        frame_id: typing.ClassVar[int] = 120  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C0078  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Nack'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

        # Class Attribute Types for unmarshaling
        _delivery_tag = 'longlong'
        _multiple = 'bit'
        _requeue = 'bit'

        def __init__(self,
                     delivery_tag: int = 0,
                     multiple: bool = False,
                     requeue: bool = True) -> None:
            """Initialize the :class:`Basic.Nack` class"""
            self.delivery_tag = delivery_tag
            self.multiple = multiple
            self.requeue = requeue

    class Properties(base.BasicProperties):
        """Content Properties

        .. Note:: The AMQP property type is named ``message_type`` as to not
                  conflict with the Python ``type`` keyword

        :param content_type: MIME content type
        :param content_encoding: MIME content encoding
        :param headers: Message header field table
        :type headers: typing.Optional[:const:`~pamqp.common.FieldTable`]
        :param delivery_mode: Non-persistent (1) or persistent (2)
        :param priority: Message priority, 0 to 9
        :param correlation_id: Application correlation identifier
        :param reply_to: Address to reply to
        :param expiration: Message expiration specification
        :param message_id: Application message identifier
        :param timestamp: Message timestamp
        :param message_type: Message type name
        :param user_id: Creating user id
        :param app_id: Creating application id
        :param cluster_id: Deprecated, must be empty
        :raises: ValueError

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Properties Attributes
            'content_type', 'content_encoding', 'headers', 'delivery_mode',
            'priority', 'correlation_id', 'reply_to', 'expiration',
            'message_id', 'timestamp', 'message_type', 'user_id', 'app_id',
            'cluster_id'
        ]

        # Flag values for marshaling / unmarshaling
        flags: typing.ClassVar[dict[str, int]] = {
            'content_type': 32768,
            'content_encoding': 16384,
            'headers': 8192,
            'delivery_mode': 4096,
            'priority': 2048,
            'correlation_id': 1024,
            'reply_to': 512,
            'expiration': 256,
            'message_id': 128,
            'timestamp': 64,
            'message_type': 32,
            'user_id': 16,
            'app_id': 8,
            'cluster_id': 4
        }

        frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x003C  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Basic.Properties'

        # Class Attribute Types for unmarshaling
        _content_type = 'shortstr'
        _content_encoding = 'shortstr'
        _headers = 'table'
        _delivery_mode = 'octet'
        _priority = 'octet'
        _correlation_id = 'shortstr'
        _reply_to = 'shortstr'
        _expiration = 'shortstr'
        _message_id = 'shortstr'
        _timestamp = 'timestamp'
        _message_type = 'shortstr'
        _user_id = 'shortstr'
        _app_id = 'shortstr'
        _cluster_id = 'shortstr'

        def __init__(self,
                     content_type: str | None = None,
                     content_encoding: str | None = None,
                     headers: common.FieldTable | None = None,
                     delivery_mode: int | None = None,
                     priority: int | None = None,
                     correlation_id: str | None = None,
                     reply_to: str | None = None,
                     expiration: str | None = None,
                     message_id: str | None = None,
                     timestamp: datetime.datetime | None = None,
                     message_type: str | None = None,
                     user_id: str | None = None,
                     app_id: str | None = None,
                     cluster_id: str = '') -> None:
            """Initialize the Basic.Properties class"""
            self.content_type = content_type
            self.content_encoding = content_encoding
            self.headers = headers
            self.delivery_mode = delivery_mode
            self.priority = priority
            self.correlation_id = correlation_id
            self.reply_to = reply_to
            self.expiration = expiration
            self.message_id = message_id
            self.timestamp = timestamp
            self.message_type = message_type
            self.user_id = user_id
            self.app_id = app_id
            self.cluster_id = cluster_id
            self.validate()

Qos

Bases: Frame

Specify quality of service

This method requests a specific quality of service. The QoS can be specified for the current channel or for all channels on the connection. The particular properties and semantics of a qos method always depend on the content class semantics. Though the qos method could in principle apply to both peers, it is currently meaningful only for the server.

Parameters:

Name	Type	Description	Default
prefetch_size	int	Prefetch window in octets - Default: 0	0
prefetch_count	int	Prefetch window in messages - Default: 0	0
global_	bool	Apply to entire connection - Default: False	False

Source code in pamqp/commands.py

class Qos(base.Frame):
    """Specify quality of service

    This method requests a specific quality of service. The QoS can be
    specified for the current channel or for all channels on the
    connection. The particular properties and semantics of a qos method
    always depend on the content class semantics. Though the qos method
    could in principle apply to both peers, it is currently meaningful only
    for the server.

    :param prefetch_size: Prefetch window in octets
        - Default: ``0``
    :param prefetch_count: Prefetch window in messages
        - Default: ``0``
    :param global_: Apply to entire connection
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'prefetch_size', 'prefetch_count', 'global_'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Qos'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Basic.QosOk']

    # Class Attribute Types for unmarshaling
    _prefetch_size = 'long'
    _prefetch_count = 'short'
    _global_ = 'bit'

    def __init__(self,
                 prefetch_size: int = 0,
                 prefetch_count: int = 0,
                 global_: bool = False) -> None:
        """Initialize the :class:`Basic.Qos` class"""
        self.prefetch_size = prefetch_size
        self.prefetch_count = prefetch_count
        self.global_ = global_

__init__(prefetch_size=0, prefetch_count=0, global_=False)

Initialize the :class:Basic.Qos class

Source code in pamqp/commands.py

def __init__(self,
             prefetch_size: int = 0,
             prefetch_count: int = 0,
             global_: bool = False) -> None:
    """Initialize the :class:`Basic.Qos` class"""
    self.prefetch_size = prefetch_size
    self.prefetch_count = prefetch_count
    self.global_ = global_

QosOk

Bases: Frame

Confirm the requested qos

This method tells the client that the requested QoS levels could be handled by the server. The requested QoS applies to all active consumers until a new QoS is defined.

Source code in pamqp/commands.py

class QosOk(base.Frame):
    """Confirm the requested qos

    This method tells the client that the requested QoS levels could be
    handled by the server. The requested QoS applies to all active
    consumers until a new QoS is defined.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.QosOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Consume

Bases: Frame

Start a queue consumer

This method asks the server to start a "consumer", which is a transient request for messages from a specific queue. Consumers last as long as the channel they were declared on, or until the client cancels them.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to consume from. - Default: ''	''
consumer_tag	str	Specifies the identifier for the consumer. The consumer tag is local to a channel, so two clients can use the same consumer tags. If this field is empty the server will generate a unique tag. - Default: ''	''
no_local	bool	Do not deliver own messages - Default: False	False
no_ack	bool	No acknowledgement needed - Default: False	False
exclusive	bool	Request exclusive access - Default: False	False
nowait	bool	Do not send a reply method - Default: False	False
arguments	Arguments | None	Arguments for declaration - Default: {}	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Consume(base.Frame):
    """Start a queue consumer

    This method asks the server to start a "consumer", which is a transient
    request for messages from a specific queue. Consumers last as long as
    the channel they were declared on, or until the client cancels them.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to consume from.
        - Default: ``''``
    :param consumer_tag: Specifies the identifier for the consumer. The
        consumer tag is local to a channel, so two clients can use the same
        consumer tags. If this field is empty the server will generate a
        unique tag.
        - Default: ``''``
    :param no_local: Do not deliver own messages
        - Default: ``False``
    :param no_ack: No acknowledgement needed
        - Default: ``False``
    :param exclusive: Request exclusive access
        - Default: ``False``
    :param nowait: Do not send a reply method
        - Default: ``False``
    :param arguments: Arguments for declaration
        - Default: ``{}``
    :type arguments: :const:`~pamqp.common.Arguments`
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'consumer_tag', 'no_local', 'no_ack',
        'exclusive', 'nowait', 'arguments'
    ]

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Consume'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Basic.ConsumeOk']

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _consumer_tag = 'shortstr'
    _no_local = 'bit'
    _no_ack = 'bit'
    _exclusive = 'bit'
    _nowait = 'bit'
    _arguments = 'table'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 consumer_tag: str = '',
                 no_local: bool = False,
                 no_ack: bool = False,
                 exclusive: bool = False,
                 nowait: bool = False,
                 arguments: common.Arguments | None = None) -> None:
        """Initialize the :class:`Basic.Consume` class"""
        self.ticket = ticket
        self.queue = queue
        self.consumer_tag = consumer_tag
        self.no_local = no_local
        self.no_ack = no_ack
        self.exclusive = exclusive
        self.nowait = nowait
        self.arguments = arguments or {}
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(ticket=0, queue='', consumer_tag='', no_local=False, no_ack=False, exclusive=False, nowait=False, arguments=None)

Initialize the :class:Basic.Consume class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             consumer_tag: str = '',
             no_local: bool = False,
             no_ack: bool = False,
             exclusive: bool = False,
             nowait: bool = False,
             arguments: common.Arguments | None = None) -> None:
    """Initialize the :class:`Basic.Consume` class"""
    self.ticket = ticket
    self.queue = queue
    self.consumer_tag = consumer_tag
    self.no_local = no_local
    self.no_ack = no_ack
    self.exclusive = exclusive
    self.nowait = nowait
    self.arguments = arguments or {}
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

ConsumeOk

Bases: Frame

Confirm a new consumer

The server provides the client with a consumer tag, which is used by the client for methods called on the consumer at a later stage.

Parameters:

Name	Type	Description	Default
consumer_tag	str | None	Holds the consumer tag specified by the client or provided by the server.	None

Source code in pamqp/commands.py

class ConsumeOk(base.Frame):
    """Confirm a new consumer

    The server provides the client with a consumer tag, which is used by
    the client for methods called on the consumer at a later stage.

    :param consumer_tag: Holds the consumer tag specified by the client or
        provided by the server.

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'consumer_tag'
    ]

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.ConsumeOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _consumer_tag = 'shortstr'

    def __init__(self, consumer_tag: str | None = None) -> None:
        """Initialize the :class:`Basic.ConsumeOk` class"""
        self.consumer_tag = consumer_tag

__init__(consumer_tag=None)

Initialize the :class:Basic.ConsumeOk class

Source code in pamqp/commands.py

def __init__(self, consumer_tag: str | None = None) -> None:
    """Initialize the :class:`Basic.ConsumeOk` class"""
    self.consumer_tag = consumer_tag

Cancel

Bases: Frame

End a queue consumer

This method cancels a consumer. This does not affect already delivered messages, but it does mean the server will not send any more messages for that consumer. The client may receive an arbitrary number of messages in between sending the cancel method and receiving the cancel- ok reply. It may also be sent from the server to the client in the event of the consumer being unexpectedly cancelled (i.e. cancelled for any reason other than the server receiving the corresponding basic.cancel from the client). This allows clients to be notified of the loss of consumers due to events such as queue deletion. Note that as it is not a MUST for clients to accept this method from the server, it is advisable for the broker to be able to identify those clients that are capable of accepting the method, through some means of capability negotiation.

Parameters:

Name	Type	Description	Default
consumer_tag	str | None	Consumer tag	None
nowait	bool	Do not send a reply method - Default: False	False

Source code in pamqp/commands.py

class Cancel(base.Frame):
    """End a queue consumer

    This method cancels a consumer. This does not affect already delivered
    messages, but it does mean the server will not send any more messages
    for that consumer. The client may receive an arbitrary number of
    messages in between sending the cancel method and receiving the cancel-
    ok reply.  It may also be sent from the server to the client in the
    event of the consumer being unexpectedly cancelled (i.e. cancelled for
    any reason other than the server receiving the corresponding
    basic.cancel from the client). This allows clients to be notified of
    the loss of consumers due to events such as queue deletion. Note that
    as it is not a MUST for clients to accept this method from the server,
    it is advisable for the broker to be able to identify those clients
    that are capable of accepting the method, through some means of
    capability negotiation.

    :param consumer_tag: Consumer tag
    :param nowait: Do not send a reply method
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'consumer_tag', 'nowait'
    ]

    frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C001E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Cancel'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Basic.CancelOk']

    # Class Attribute Types for unmarshaling
    _consumer_tag = 'shortstr'
    _nowait = 'bit'

    def __init__(self,
                 consumer_tag: str | None = None,
                 nowait: bool = False) -> None:
        """Initialize the :class:`Basic.Cancel` class"""
        self.consumer_tag = consumer_tag
        self.nowait = nowait if nowait is not None else False

__init__(consumer_tag=None, nowait=False)

Initialize the :class:Basic.Cancel class

Source code in pamqp/commands.py

def __init__(self,
             consumer_tag: str | None = None,
             nowait: bool = False) -> None:
    """Initialize the :class:`Basic.Cancel` class"""
    self.consumer_tag = consumer_tag
    self.nowait = nowait if nowait is not None else False

CancelOk

Bases: Frame

Confirm a cancelled consumer

This method confirms that the cancellation was completed.

Parameters:

Name	Type	Description	Default
consumer_tag	str | None	Consumer tag	None

Source code in pamqp/commands.py

class CancelOk(base.Frame):
    """Confirm a cancelled consumer

    This method confirms that the cancellation was completed.

    :param consumer_tag: Consumer tag

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'consumer_tag'
    ]

    frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C001F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.CancelOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _consumer_tag = 'shortstr'

    def __init__(self, consumer_tag: str | None = None) -> None:
        """Initialize the :class:`Basic.CancelOk` class"""
        self.consumer_tag = consumer_tag

__init__(consumer_tag=None)

Initialize the :class:Basic.CancelOk class

Source code in pamqp/commands.py

def __init__(self, consumer_tag: str | None = None) -> None:
    """Initialize the :class:`Basic.CancelOk` class"""
    self.consumer_tag = consumer_tag

Publish

Bases: Frame

Publish a message

This method publishes a message to a specific exchange. The message will be routed to queues as defined by the exchange configuration and distributed to any active consumers when the transaction, if any, is committed.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
exchange	str	Specifies the name of the exchange to publish to. The exchange name can be empty, meaning the default exchange. If the exchange name is specified, and that exchange does not exist, the server will raise a channel exception. - Default: ''	''
routing_key	str	Message routing key - Default: ''	''
mandatory	bool	Indicate mandatory routing - Default: False	False
immediate	bool	Request immediate delivery - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Publish(base.Frame):
    """Publish a message

    This method publishes a message to a specific exchange. The message
    will be routed to queues as defined by the exchange configuration and
    distributed to any active consumers when the transaction, if any, is
    committed.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param exchange: Specifies the name of the exchange to publish to. The
        exchange name can be empty, meaning the default exchange. If the
        exchange name is specified, and that exchange does not exist, the
        server will raise a channel exception.
        - Default: ``''``
    :param routing_key: Message routing key
        - Default: ``''``
    :param mandatory: Indicate mandatory routing
        - Default: ``False``
    :param immediate: Request immediate delivery
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'exchange', 'routing_key', 'mandatory', 'immediate'
    ]

    frame_id: typing.ClassVar[int] = 40  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0028  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Publish'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'
    _mandatory = 'bit'
    _immediate = 'bit'

    def __init__(self,
                 ticket: int = 0,
                 exchange: str = '',
                 routing_key: str = '',
                 mandatory: bool = False,
                 immediate: bool = False) -> None:
        """Initialize the :class:`Basic.Publish` class"""
        self.ticket = ticket
        self.exchange = exchange
        self.routing_key = routing_key
        self.mandatory = mandatory
        self.immediate = immediate
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(ticket=0, exchange='', routing_key='', mandatory=False, immediate=False)

Initialize the :class:Basic.Publish class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             exchange: str = '',
             routing_key: str = '',
             mandatory: bool = False,
             immediate: bool = False) -> None:
    """Initialize the :class:`Basic.Publish` class"""
    self.ticket = ticket
    self.exchange = exchange
    self.routing_key = routing_key
    self.mandatory = mandatory
    self.immediate = immediate
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

Return

Bases: Frame

Return a failed message

This method returns an undeliverable message that was published with the "immediate" flag set, or an unroutable message published with the "mandatory" flag set. The reply code and text provide information about the reason that the message was undeliverable.

Parameters:

Name	Type	Description	Default
reply_code	int | None	Reply code from server	None
reply_text	str	Localised reply text - Default: ''	''
exchange	str	Specifies the name of the exchange that the message was originally published to. May be empty, meaning the default exchange. - Default: ''	''
routing_key	str | None	Message routing key	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Return(base.Frame):
    """Return a failed message

    This method returns an undeliverable message that was published with
    the "immediate" flag set, or an unroutable message published with the
    "mandatory" flag set. The reply code and text provide information about
    the reason that the message was undeliverable.

    :param reply_code: Reply code from server
    :param reply_text: Localised reply text
        - Default: ``''``
    :param exchange: Specifies the name of the exchange that the message
        was originally published to.  May be empty, meaning the default
        exchange.
        - Default: ``''``
    :param routing_key: Message routing key
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'reply_code', 'reply_text', 'exchange', 'routing_key'
    ]

    frame_id: typing.ClassVar[int] = 50  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0032  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Return'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _reply_code = 'short'
    _reply_text = 'shortstr'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'

    def __init__(self,
                 reply_code: int | None = None,
                 reply_text: str = '',
                 exchange: str = '',
                 routing_key: str | None = None) -> None:
        """Initialize the :class:`Basic.Return` class"""
        self.reply_code = reply_code
        self.reply_text = reply_text or ''
        self.exchange = exchange or ''
        self.routing_key = routing_key
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(reply_code=None, reply_text='', exchange='', routing_key=None)

Initialize the :class:Basic.Return class

Source code in pamqp/commands.py

def __init__(self,
             reply_code: int | None = None,
             reply_text: str = '',
             exchange: str = '',
             routing_key: str | None = None) -> None:
    """Initialize the :class:`Basic.Return` class"""
    self.reply_code = reply_code
    self.reply_text = reply_text or ''
    self.exchange = exchange or ''
    self.routing_key = routing_key
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

Deliver

Bases: Frame

Notify the client of a consumer message

This method delivers a message to the client, via a consumer. In the asynchronous message delivery model, the client starts a consumer using the Consume method, then the server responds with Deliver methods as and when messages arrive for that consumer.

Parameters:

Name	Type	Description	Default
consumer_tag	str | None	Consumer tag	None
delivery_tag	int | None	Server-assigned delivery tag	None
redelivered	bool	Message is being redelivered - Default: False	False
exchange	str	Specifies the name of the exchange that the message was originally published to. May be empty, indicating the default exchange. - Default: ''	''
routing_key	str | None	Message routing key	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Deliver(base.Frame):
    """Notify the client of a consumer message

    This method delivers a message to the client, via a consumer. In the
    asynchronous message delivery model, the client starts a consumer using
    the Consume method, then the server responds with Deliver methods as
    and when messages arrive for that consumer.

    :param consumer_tag: Consumer tag
    :param delivery_tag: Server-assigned delivery tag
    :param redelivered: Message is being redelivered
        - Default: ``False``
    :param exchange: Specifies the name of the exchange that the message
        was originally published to. May be empty, indicating the default
        exchange.
        - Default: ``''``
    :param routing_key: Message routing key
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'consumer_tag', 'delivery_tag', 'redelivered', 'exchange',
        'routing_key'
    ]

    frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C003C  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Deliver'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _consumer_tag = 'shortstr'
    _delivery_tag = 'longlong'
    _redelivered = 'bit'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'

    def __init__(self,
                 consumer_tag: str | None = None,
                 delivery_tag: int | None = None,
                 redelivered: bool = False,
                 exchange: str = '',
                 routing_key: str | None = None) -> None:
        """Initialize the :class:`Basic.Deliver` class"""
        self.consumer_tag = consumer_tag
        self.delivery_tag = delivery_tag
        self.redelivered = redelivered if redelivered is not None else False
        self.exchange = exchange or ''
        self.routing_key = routing_key
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(consumer_tag=None, delivery_tag=None, redelivered=False, exchange='', routing_key=None)

Initialize the :class:Basic.Deliver class

Source code in pamqp/commands.py

def __init__(self,
             consumer_tag: str | None = None,
             delivery_tag: int | None = None,
             redelivered: bool = False,
             exchange: str = '',
             routing_key: str | None = None) -> None:
    """Initialize the :class:`Basic.Deliver` class"""
    self.consumer_tag = consumer_tag
    self.delivery_tag = delivery_tag
    self.redelivered = redelivered if redelivered is not None else False
    self.exchange = exchange or ''
    self.routing_key = routing_key
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

Get

Bases: Frame

Direct access to a queue

This method provides a direct access to the messages in a queue using a synchronous dialogue that is designed for specific types of application where synchronous functionality is more important than performance.

Parameters:

Name	Type	Description	Default
ticket	int	Deprecated, must be 0 - Default: 0	0
queue	str	Specifies the name of the queue to get a message from. - Default: ''	''
no_ack	bool	No acknowledgement needed - Default: False	False

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class Get(base.Frame):
    """Direct access to a queue

    This method provides a direct access to the messages in a queue using a
    synchronous dialogue that is designed for specific types of application
    where synchronous functionality is more important than performance.

    :param ticket: Deprecated, must be ``0``
        - Default: ``0``
    :param queue: Specifies the name of the queue to get a message from.
        - Default: ``''``
    :param no_ack: No acknowledgement needed
        - Default: ``False``
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'ticket', 'queue', 'no_ack'
    ]

    frame_id: typing.ClassVar[int] = 70  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0046  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Get'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = [
        'Basic.GetOk', 'Basic.GetEmpty'
    ]

    # Class Attribute Types for unmarshaling
    _ticket = 'short'
    _queue = 'shortstr'
    _no_ack = 'bit'

    def __init__(self,
                 ticket: int = 0,
                 queue: str = '',
                 no_ack: bool = False) -> None:
        """Initialize the :class:`Basic.Get` class"""
        self.ticket = ticket
        self.queue = queue
        self.no_ack = no_ack
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.ticket is not None and self.ticket != 0:
            raise ValueError('ticket must be 0')
        if self.queue is not None and len(self.queue) > 256:
            raise ValueError('Max length exceeded for queue')
        if self.queue is not None and not constants.DOMAIN_REGEX[
                'queue-name'].fullmatch(self.queue):
            raise ValueError('Invalid value for queue')

__init__(ticket=0, queue='', no_ack=False)

Initialize the :class:Basic.Get class

Source code in pamqp/commands.py

def __init__(self,
             ticket: int = 0,
             queue: str = '',
             no_ack: bool = False) -> None:
    """Initialize the :class:`Basic.Get` class"""
    self.ticket = ticket
    self.queue = queue
    self.no_ack = no_ack
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.ticket is not None and self.ticket != 0:
        raise ValueError('ticket must be 0')
    if self.queue is not None and len(self.queue) > 256:
        raise ValueError('Max length exceeded for queue')
    if self.queue is not None and not constants.DOMAIN_REGEX[
            'queue-name'].fullmatch(self.queue):
        raise ValueError('Invalid value for queue')

GetOk

Bases: Frame

Provide client with a message

This method delivers a message to the client following a get method. A message delivered by 'get-ok' must be acknowledged unless the no-ack option was set in the get method.

Parameters:

Name	Type	Description	Default
delivery_tag	int | None	Server-assigned delivery tag	None
redelivered	bool	Message is being redelivered - Default: False	False
exchange	str	Specifies the name of the exchange that the message was originally published to. If empty, the message was published to the default exchange. - Default: ''	''
routing_key	str | None	Message routing key	None
message_count	int | None	Number of messages in queue	None

Raises:

Type	Description
ValueError	when an argument fails to validate

Source code in pamqp/commands.py

class GetOk(base.Frame):
    """Provide client with a message

    This method delivers a message to the client following a get method. A
    message delivered by 'get-ok' must be acknowledged unless the no-ack
    option was set in the get method.

    :param delivery_tag: Server-assigned delivery tag
    :param redelivered: Message is being redelivered
        - Default: ``False``
    :param exchange: Specifies the name of the exchange that the message
        was originally published to. If empty, the message was published to
        the default exchange.
        - Default: ``''``
    :param routing_key: Message routing key
    :param message_count: Number of messages in queue
    :raises ValueError: when an argument fails to validate

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'delivery_tag', 'redelivered', 'exchange', 'routing_key',
        'message_count'
    ]

    frame_id: typing.ClassVar[int] = 71  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0047  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.GetOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _delivery_tag = 'longlong'
    _redelivered = 'bit'
    _exchange = 'shortstr'
    _routing_key = 'shortstr'
    _message_count = 'long'

    def __init__(self,
                 delivery_tag: int | None = None,
                 redelivered: bool = False,
                 exchange: str = '',
                 routing_key: str | None = None,
                 message_count: int | None = None) -> None:
        """Initialize the :class:`Basic.GetOk` class"""
        self.delivery_tag = delivery_tag
        self.redelivered = redelivered if redelivered is not None else False
        self.exchange = exchange or ''
        self.routing_key = routing_key
        self.message_count = message_count
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.exchange is not None and len(self.exchange) > 127:
            raise ValueError('Max length exceeded for exchange')
        if self.exchange is not None and not constants.DOMAIN_REGEX[
                'exchange-name'].fullmatch(self.exchange):
            raise ValueError('Invalid value for exchange')

__init__(delivery_tag=None, redelivered=False, exchange='', routing_key=None, message_count=None)

Initialize the :class:Basic.GetOk class

Source code in pamqp/commands.py

def __init__(self,
             delivery_tag: int | None = None,
             redelivered: bool = False,
             exchange: str = '',
             routing_key: str | None = None,
             message_count: int | None = None) -> None:
    """Initialize the :class:`Basic.GetOk` class"""
    self.delivery_tag = delivery_tag
    self.redelivered = redelivered if redelivered is not None else False
    self.exchange = exchange or ''
    self.routing_key = routing_key
    self.message_count = message_count
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.exchange is not None and len(self.exchange) > 127:
        raise ValueError('Max length exceeded for exchange')
    if self.exchange is not None and not constants.DOMAIN_REGEX[
            'exchange-name'].fullmatch(self.exchange):
        raise ValueError('Invalid value for exchange')

GetEmpty

Bases: Frame

Indicate no messages available

This method tells the client that the queue has no messages available for the client.

Parameters:

Name	Type	Description	Default
cluster_id	str	Deprecated, must be empty - Default: ''	''

Source code in pamqp/commands.py

class GetEmpty(base.Frame):
    """Indicate no messages available

    This method tells the client that the queue has no messages available
    for the client.

    :param cluster_id: Deprecated, must be empty
        - Default: ``''``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'cluster_id'
    ]

    frame_id: typing.ClassVar[int] = 72  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0048  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.GetEmpty'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _cluster_id = 'shortstr'

    def __init__(self, cluster_id: str = '') -> None:
        """Initialize the :class:`Basic.GetEmpty` class"""
        self.cluster_id = cluster_id
        self.validate()

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

        :raises ValueError: on validation error

        """
        if self.cluster_id is not None and self.cluster_id != '':
            raise ValueError('cluster_id must be empty')

__init__(cluster_id='')

Initialize the :class:Basic.GetEmpty class

Source code in pamqp/commands.py

def __init__(self, cluster_id: str = '') -> None:
    """Initialize the :class:`Basic.GetEmpty` class"""
    self.cluster_id = cluster_id
    self.validate()

validate()

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

Raises:

Type	Description
ValueError	on validation error

Source code in pamqp/commands.py

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

    :raises ValueError: on validation error

    """
    if self.cluster_id is not None and self.cluster_id != '':
        raise ValueError('cluster_id must be empty')

Ack

Bases: Frame

Acknowledge one or more messages

When sent by the client, this method acknowledges one or more messages delivered via the Deliver or Get-Ok methods. When sent by server, this method acknowledges one or more messages published with the Publish method on a channel in confirm mode. The acknowledgement can be for a single message or a set of messages up to and including a specific message.

Parameters:

Name	Type	Description	Default
delivery_tag	int	Server-assigned delivery tag - Default: 0	0
multiple	bool	Acknowledge multiple messages - Default: False	False

Source code in pamqp/commands.py

class Ack(base.Frame):
    """Acknowledge one or more messages

    When sent by the client, this method acknowledges one or more messages
    delivered via the Deliver or Get-Ok methods.  When sent by server, this
    method acknowledges one or more messages published with the Publish
    method on a channel in confirm mode.  The acknowledgement can be for a
    single message or a set of messages up to and including a specific
    message.

    :param delivery_tag: Server-assigned delivery tag
        - Default: ``0``
    :param multiple: Acknowledge multiple messages
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'delivery_tag', 'multiple'
    ]

    frame_id: typing.ClassVar[int] = 80  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0050  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Ack'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _delivery_tag = 'longlong'
    _multiple = 'bit'

    def __init__(self,
                 delivery_tag: int = 0,
                 multiple: bool = False) -> None:
        """Initialize the :class:`Basic.Ack` class"""
        self.delivery_tag = delivery_tag
        self.multiple = multiple

__init__(delivery_tag=0, multiple=False)

Initialize the :class:Basic.Ack class

Source code in pamqp/commands.py

def __init__(self,
             delivery_tag: int = 0,
             multiple: bool = False) -> None:
    """Initialize the :class:`Basic.Ack` class"""
    self.delivery_tag = delivery_tag
    self.multiple = multiple

Reject

Bases: Frame

Reject an incoming message

This method allows a client to reject a message. It can be used to interrupt and cancel large incoming messages, or return untreatable messages to their original queue.

Parameters:

Name	Type	Description	Default
delivery_tag	int | None	Server-assigned delivery tag	None
requeue	bool	Requeue the message - Default: True	True

Source code in pamqp/commands.py

class Reject(base.Frame):
    """Reject an incoming message

    This method allows a client to reject a message. It can be used to
    interrupt and cancel large incoming messages, or return untreatable
    messages to their original queue.

    :param delivery_tag: Server-assigned delivery tag
    :param requeue: Requeue the message
        - Default: ``True``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'delivery_tag', 'requeue'
    ]

    frame_id: typing.ClassVar[int] = 90  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C005A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Reject'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _delivery_tag = 'longlong'
    _requeue = 'bit'

    def __init__(self,
                 delivery_tag: int | None = None,
                 requeue: bool = True) -> None:
        """Initialize the :class:`Basic.Reject` class"""
        self.delivery_tag = delivery_tag
        self.requeue = requeue if requeue is not None else True

__init__(delivery_tag=None, requeue=True)

Initialize the :class:Basic.Reject class

Source code in pamqp/commands.py

def __init__(self,
             delivery_tag: int | None = None,
             requeue: bool = True) -> None:
    """Initialize the :class:`Basic.Reject` class"""
    self.delivery_tag = delivery_tag
    self.requeue = requeue if requeue is not None else True

RecoverAsync

Bases: Frame

Redeliver unacknowledged messages

This method asks the server to redeliver all unacknowledged messages on a specified channel. Zero or more messages may be redelivered. This method is deprecated in favour of the synchronous Recover/Recover-Ok.

.. deprecated:: This command is deprecated in AMQP 0-9-1

Parameters:

Name	Type	Description	Default
requeue	bool	Requeue the message - Default: False	False

Source code in pamqp/commands.py

class RecoverAsync(base.Frame):
    """Redeliver unacknowledged messages

    This method asks the server to redeliver all unacknowledged messages on
    a specified channel. Zero or more messages may be redelivered.  This
    method is deprecated in favour of the synchronous Recover/Recover-Ok.

    .. deprecated:: This command is deprecated in AMQP 0-9-1

    :param requeue: Requeue the message
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'requeue'
    ]

    frame_id: typing.ClassVar[int] = 100  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0064  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.RecoverAsync'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _requeue = 'bit'

    def __init__(self, requeue: bool = False) -> None:
        """Initialize the :class:`Basic.RecoverAsync` class"""
        self.requeue = requeue
        warnings.warn(constants.DEPRECATION_WARNING,
                      category=DeprecationWarning)

__init__(requeue=False)

Initialize the :class:Basic.RecoverAsync class

Source code in pamqp/commands.py

def __init__(self, requeue: bool = False) -> None:
    """Initialize the :class:`Basic.RecoverAsync` class"""
    self.requeue = requeue
    warnings.warn(constants.DEPRECATION_WARNING,
                  category=DeprecationWarning)

Recover

Bases: Frame

Redeliver unacknowledged messages

This method asks the server to redeliver all unacknowledged messages on a specified channel. Zero or more messages may be redelivered. This method replaces the asynchronous Recover.

Parameters:

Name	Type	Description	Default
requeue	bool	Requeue the message - Default: False	False

Source code in pamqp/commands.py

class Recover(base.Frame):
    """Redeliver unacknowledged messages

    This method asks the server to redeliver all unacknowledged messages on
    a specified channel. Zero or more messages may be redelivered.  This
    method replaces the asynchronous Recover.

    :param requeue: Requeue the message
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'requeue'
    ]

    frame_id: typing.ClassVar[int] = 110  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C006E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Recover'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Basic.RecoverOk']

    # Class Attribute Types for unmarshaling
    _requeue = 'bit'

    def __init__(self, requeue: bool = False) -> None:
        """Initialize the :class:`Basic.Recover` class"""
        self.requeue = requeue

__init__(requeue=False)

Initialize the :class:Basic.Recover class

Source code in pamqp/commands.py

def __init__(self, requeue: bool = False) -> None:
    """Initialize the :class:`Basic.Recover` class"""
    self.requeue = requeue

RecoverOk

Bases: Frame

Confirm recovery

This method acknowledges a :class:Basic.Recover method.

Source code in pamqp/commands.py

class RecoverOk(base.Frame):
    """Confirm recovery

    This method acknowledges a :class:`Basic.Recover` method.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 111  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C006F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.RecoverOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Nack

Bases: Frame

Reject one or more incoming messages

This method allows a client to reject one or more incoming messages. It can be used to interrupt and cancel large incoming messages, or return untreatable messages to their original queue. This method is also used by the server to inform publishers on channels in confirm mode of unhandled messages. If a publisher receives this method, it probably needs to republish the offending messages.

Parameters:

Name	Type	Description	Default
delivery_tag	int	Server-assigned delivery tag - Default: 0	0
multiple	bool	Reject multiple messages - Default: False	False
requeue	bool	Requeue the message - Default: True	True

Source code in pamqp/commands.py

class Nack(base.Frame):
    """Reject one or more incoming messages

    This method allows a client to reject one or more incoming messages. It
    can be used to interrupt and cancel large incoming messages, or return
    untreatable messages to their original queue.  This method is also used
    by the server to inform publishers on channels in confirm mode of
    unhandled messages.  If a publisher receives this method, it probably
    needs to republish the offending messages.

    :param delivery_tag: Server-assigned delivery tag
        - Default: ``0``
    :param multiple: Reject multiple messages
        - Default: ``False``
    :param requeue: Requeue the message
        - Default: ``True``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'delivery_tag', 'multiple', 'requeue'
    ]

    frame_id: typing.ClassVar[int] = 120  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C0078  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Nack'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

    # Class Attribute Types for unmarshaling
    _delivery_tag = 'longlong'
    _multiple = 'bit'
    _requeue = 'bit'

    def __init__(self,
                 delivery_tag: int = 0,
                 multiple: bool = False,
                 requeue: bool = True) -> None:
        """Initialize the :class:`Basic.Nack` class"""
        self.delivery_tag = delivery_tag
        self.multiple = multiple
        self.requeue = requeue

__init__(delivery_tag=0, multiple=False, requeue=True)

Initialize the :class:Basic.Nack class

Source code in pamqp/commands.py

def __init__(self,
             delivery_tag: int = 0,
             multiple: bool = False,
             requeue: bool = True) -> None:
    """Initialize the :class:`Basic.Nack` class"""
    self.delivery_tag = delivery_tag
    self.multiple = multiple
    self.requeue = requeue

Properties

Bases: BasicProperties

Content Properties

.. Note:: The AMQP property type is named message_type as to not conflict with the Python type keyword

Parameters:

Name	Type	Description	Default
content_type	str | None	MIME content type	None
content_encoding	str | None	MIME content encoding	None
headers	FieldTable | None	Message header field table	None
delivery_mode	int | None	Non-persistent (1) or persistent (2)	None
priority	int | None	Message priority, 0 to 9	None
correlation_id	str | None	Application correlation identifier	None
reply_to	str | None	Address to reply to	None
expiration	str | None	Message expiration specification	None
message_id	str | None	Application message identifier	None
timestamp	datetime | None	Message timestamp	None
message_type	str | None	Message type name	None
user_id	str | None	Creating user id	None
app_id	str | None	Creating application id	None
cluster_id	str	Deprecated, must be empty	''

Source code in pamqp/commands.py

class Properties(base.BasicProperties):
    """Content Properties

    .. Note:: The AMQP property type is named ``message_type`` as to not
              conflict with the Python ``type`` keyword

    :param content_type: MIME content type
    :param content_encoding: MIME content encoding
    :param headers: Message header field table
    :type headers: typing.Optional[:const:`~pamqp.common.FieldTable`]
    :param delivery_mode: Non-persistent (1) or persistent (2)
    :param priority: Message priority, 0 to 9
    :param correlation_id: Application correlation identifier
    :param reply_to: Address to reply to
    :param expiration: Message expiration specification
    :param message_id: Application message identifier
    :param timestamp: Message timestamp
    :param message_type: Message type name
    :param user_id: Creating user id
    :param app_id: Creating application id
    :param cluster_id: Deprecated, must be empty
    :raises: ValueError

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Properties Attributes
        'content_type', 'content_encoding', 'headers', 'delivery_mode',
        'priority', 'correlation_id', 'reply_to', 'expiration',
        'message_id', 'timestamp', 'message_type', 'user_id', 'app_id',
        'cluster_id'
    ]

    # Flag values for marshaling / unmarshaling
    flags: typing.ClassVar[dict[str, int]] = {
        'content_type': 32768,
        'content_encoding': 16384,
        'headers': 8192,
        'delivery_mode': 4096,
        'priority': 2048,
        'correlation_id': 1024,
        'reply_to': 512,
        'expiration': 256,
        'message_id': 128,
        'timestamp': 64,
        'message_type': 32,
        'user_id': 16,
        'app_id': 8,
        'cluster_id': 4
    }

    frame_id: typing.ClassVar[int] = 60  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x003C  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Basic.Properties'

    # Class Attribute Types for unmarshaling
    _content_type = 'shortstr'
    _content_encoding = 'shortstr'
    _headers = 'table'
    _delivery_mode = 'octet'
    _priority = 'octet'
    _correlation_id = 'shortstr'
    _reply_to = 'shortstr'
    _expiration = 'shortstr'
    _message_id = 'shortstr'
    _timestamp = 'timestamp'
    _message_type = 'shortstr'
    _user_id = 'shortstr'
    _app_id = 'shortstr'
    _cluster_id = 'shortstr'

    def __init__(self,
                 content_type: str | None = None,
                 content_encoding: str | None = None,
                 headers: common.FieldTable | None = None,
                 delivery_mode: int | None = None,
                 priority: int | None = None,
                 correlation_id: str | None = None,
                 reply_to: str | None = None,
                 expiration: str | None = None,
                 message_id: str | None = None,
                 timestamp: datetime.datetime | None = None,
                 message_type: str | None = None,
                 user_id: str | None = None,
                 app_id: str | None = None,
                 cluster_id: str = '') -> None:
        """Initialize the Basic.Properties class"""
        self.content_type = content_type
        self.content_encoding = content_encoding
        self.headers = headers
        self.delivery_mode = delivery_mode
        self.priority = priority
        self.correlation_id = correlation_id
        self.reply_to = reply_to
        self.expiration = expiration
        self.message_id = message_id
        self.timestamp = timestamp
        self.message_type = message_type
        self.user_id = user_id
        self.app_id = app_id
        self.cluster_id = cluster_id
        self.validate()

__init__(content_type=None, content_encoding=None, headers=None, delivery_mode=None, priority=None, correlation_id=None, reply_to=None, expiration=None, message_id=None, timestamp=None, message_type=None, user_id=None, app_id=None, cluster_id='')

Initialize the Basic.Properties class

Source code in pamqp/commands.py

def __init__(self,
             content_type: str | None = None,
             content_encoding: str | None = None,
             headers: common.FieldTable | None = None,
             delivery_mode: int | None = None,
             priority: int | None = None,
             correlation_id: str | None = None,
             reply_to: str | None = None,
             expiration: str | None = None,
             message_id: str | None = None,
             timestamp: datetime.datetime | None = None,
             message_type: str | None = None,
             user_id: str | None = None,
             app_id: str | None = None,
             cluster_id: str = '') -> None:
    """Initialize the Basic.Properties class"""
    self.content_type = content_type
    self.content_encoding = content_encoding
    self.headers = headers
    self.delivery_mode = delivery_mode
    self.priority = priority
    self.correlation_id = correlation_id
    self.reply_to = reply_to
    self.expiration = expiration
    self.message_id = message_id
    self.timestamp = timestamp
    self.message_type = message_type
    self.user_id = user_id
    self.app_id = app_id
    self.cluster_id = cluster_id
    self.validate()

Tx

Work with transactions

The Tx class allows publish and ack operations to be batched into atomic units of work. The intention is that all publish and ack requests issued within a transaction will complete successfully or none of them will. Servers SHOULD implement atomic transactions at least where all publish or ack requests affect a single queue. Transactions that cover multiple queues may be non-atomic, given that queues can be created and destroyed asynchronously, and such events do not form part of any transaction. Further, the behaviour of transactions with respect to the immediate and mandatory flags on :class:Basic.Publish methods is not defined.

Source code in pamqp/commands.py

class Tx:
    """Work with transactions

    The Tx class allows publish and ack operations to be batched into atomic
    units of work.  The intention is that all publish and ack requests issued
    within a transaction will complete successfully or none of them will.
    Servers SHOULD implement atomic transactions at least where all publish or
    ack requests affect a single queue.  Transactions that cover multiple
    queues may be non-atomic, given that queues can be created and destroyed
    asynchronously, and such events do not form part of any transaction.
    Further, the behaviour of transactions with respect to the immediate and
    mandatory flags on :class:`Basic.Publish` methods is not defined.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 90  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A0000  # pamqp Mapping Index

    class Select(base.Frame):
        """Select standard transaction mode

        This method sets the channel to use standard transactions. The client
        must use this method at least once on a channel before using the Commit
        or Rollback methods.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.Select'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Tx.SelectOk']

    class SelectOk(base.Frame):
        """Confirm transaction mode

        This method confirms to the client that the channel was successfully
        set to use standard transactions.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.SelectOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Commit(base.Frame):
        """Commit the current transaction

        This method commits all message publications and acknowledgments
        performed in the current transaction.  A new transaction starts
        immediately after a commit.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A0014  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.Commit'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Tx.CommitOk']

    class CommitOk(base.Frame):
        """Confirm a successful commit

        This method confirms to the client that the commit succeeded. Note that
        if a commit fails, the server raises a channel exception.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A0015  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.CommitOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

    class Rollback(base.Frame):
        """Abandon the current transaction

        This method abandons all message publications and acknowledgments
        performed in the current transaction. A new transaction starts
        immediately after a rollback. Note that unacked messages will not be
        automatically redelivered by rollback; if that is required an explicit
        recover call should be issued.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A001E  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.Rollback'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Tx.RollbackOk']

    class RollbackOk(base.Frame):
        """Confirm successful rollback

        This method confirms to the client that the rollback succeeded. Note
        that if an rollback fails, the server raises a channel exception.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x005A001F  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Tx.RollbackOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Select

Bases: Frame

Select standard transaction mode

This method sets the channel to use standard transactions. The client must use this method at least once on a channel before using the Commit or Rollback methods.

Source code in pamqp/commands.py

class Select(base.Frame):
    """Select standard transaction mode

    This method sets the channel to use standard transactions. The client
    must use this method at least once on a channel before using the Commit
    or Rollback methods.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.Select'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Tx.SelectOk']

SelectOk

Bases: Frame

Confirm transaction mode

This method confirms to the client that the channel was successfully set to use standard transactions.

Source code in pamqp/commands.py

class SelectOk(base.Frame):
    """Confirm transaction mode

    This method confirms to the client that the channel was successfully
    set to use standard transactions.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.SelectOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Commit

Bases: Frame

Commit the current transaction

This method commits all message publications and acknowledgments performed in the current transaction. A new transaction starts immediately after a commit.

Source code in pamqp/commands.py

class Commit(base.Frame):
    """Commit the current transaction

    This method commits all message publications and acknowledgments
    performed in the current transaction.  A new transaction starts
    immediately after a commit.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 20  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A0014  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.Commit'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Tx.CommitOk']

CommitOk

Bases: Frame

Confirm a successful commit

This method confirms to the client that the commit succeeded. Note that if a commit fails, the server raises a channel exception.

Source code in pamqp/commands.py

class CommitOk(base.Frame):
    """Confirm a successful commit

    This method confirms to the client that the commit succeeded. Note that
    if a commit fails, the server raises a channel exception.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 21  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A0015  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.CommitOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Rollback

Bases: Frame

Abandon the current transaction

This method abandons all message publications and acknowledgments performed in the current transaction. A new transaction starts immediately after a rollback. Note that unacked messages will not be automatically redelivered by rollback; if that is required an explicit recover call should be issued.

Source code in pamqp/commands.py

class Rollback(base.Frame):
    """Abandon the current transaction

    This method abandons all message publications and acknowledgments
    performed in the current transaction. A new transaction starts
    immediately after a rollback. Note that unacked messages will not be
    automatically redelivered by rollback; if that is required an explicit
    recover call should be issued.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 30  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A001E  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.Rollback'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Tx.RollbackOk']

RollbackOk

Bases: Frame

Confirm successful rollback

This method confirms to the client that the rollback succeeded. Note that if an rollback fails, the server raises a channel exception.

Source code in pamqp/commands.py

class RollbackOk(base.Frame):
    """Confirm successful rollback

    This method confirms to the client that the rollback succeeded. Note
    that if an rollback fails, the server raises a channel exception.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 31  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x005A001F  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Tx.RollbackOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method

Confirm

Work with confirms

The Confirm class allows publishers to put the channel in confirm mode and subsequently be notified when messages have been handled by the broker. The intention is that all messages published on a channel in confirm mode will be acknowledged at some point. By acknowledging a message the broker assumes responsibility for it and indicates that it has done something it deems reasonable with it. Unroutable mandatory or immediate messages are acknowledged right after the :class:Basic.Return method. Messages are acknowledged when all queues to which the message has been routed have either delivered the message and received an acknowledgement (if required), or enqueued the message (and persisted it if required). Published messages are assigned ascending sequence numbers, starting at 1 with the first :class:Confirm.Select method. The server confirms messages by sending :class:Basic.Ack methods referring to these sequence numbers.

Source code in pamqp/commands.py

class Confirm:
    """Work with confirms

    The Confirm class allows publishers to put the channel in confirm mode and
    subsequently be notified when messages have been handled by the broker.
    The intention is that all messages published on a channel in confirm mode
    will be acknowledged at some point.  By acknowledging a message the broker
    assumes responsibility for it and indicates that it has done something it
    deems reasonable with it.  Unroutable mandatory or immediate messages are
    acknowledged right after the :class:`Basic.Return` method. Messages are
    acknowledged when all queues to which the message has been routed have
    either delivered the message and received an acknowledgement (if required),
    or enqueued the message (and persisted it if required).  Published messages
    are assigned ascending sequence numbers, starting at 1 with the first
    :class:`Confirm.Select` method. The server confirms messages by sending
    :class:`Basic.Ack` methods referring to these sequence numbers.

    """
    __slots__: typing.ClassVar[list[str]] = []

    frame_id: typing.ClassVar[int] = 85  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x00550000  # pamqp Mapping Index

    class Select(base.Frame):
        """Select confirm mode (i.e. enable publisher acknowledgements)

        This method sets the channel to use publisher acknowledgements. The
        client can only use this method on a non-transactional channel.

        :param nowait: Do not send a reply method
            - Default: ``False``

        """
        __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
            'nowait'
        ]

        frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0055000A  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Confirm.Select'
        synchronous: typing.ClassVar[
            bool] = True  # Indicates if this is a synchronous AMQP method
        # Valid responses to this method
        valid_responses: typing.ClassVar[list[str]] = ['Confirm.SelectOk']

        # Class Attribute Types for unmarshaling
        _nowait = 'bit'

        def __init__(self, nowait: bool = False) -> None:
            """Initialize the :class:`Confirm.Select` class"""
            self.nowait = nowait

    class SelectOk(base.Frame):
        """Acknowledge confirm mode

        This method confirms to the client that the channel was successfully
        set to use publisher acknowledgements.

        """
        __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

        frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
        index: typing.ClassVar[int] = 0x0055000B  # pamqp Mapping Index
        name: typing.ClassVar[str] = 'Confirm.SelectOk'
        synchronous: typing.ClassVar[
            bool] = False  # Indicates if this is a synchronous AMQP method

Select

Bases: Frame

Select confirm mode (i.e. enable publisher acknowledgements)

This method sets the channel to use publisher acknowledgements. The client can only use this method on a non-transactional channel.

Parameters:

Name	Type	Description	Default
nowait	bool	Do not send a reply method - Default: False	False

Source code in pamqp/commands.py

class Select(base.Frame):
    """Select confirm mode (i.e. enable publisher acknowledgements)

    This method sets the channel to use publisher acknowledgements. The
    client can only use this method on a non-transactional channel.

    :param nowait: Do not send a reply method
        - Default: ``False``

    """
    __slots__: typing.ClassVar[list[str]] = [  # AMQ Method Attributes
        'nowait'
    ]

    frame_id: typing.ClassVar[int] = 10  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0055000A  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Confirm.Select'
    synchronous: typing.ClassVar[
        bool] = True  # Indicates if this is a synchronous AMQP method
    # Valid responses to this method
    valid_responses: typing.ClassVar[list[str]] = ['Confirm.SelectOk']

    # Class Attribute Types for unmarshaling
    _nowait = 'bit'

    def __init__(self, nowait: bool = False) -> None:
        """Initialize the :class:`Confirm.Select` class"""
        self.nowait = nowait

__init__(nowait=False)

Initialize the :class:Confirm.Select class

Source code in pamqp/commands.py

def __init__(self, nowait: bool = False) -> None:
    """Initialize the :class:`Confirm.Select` class"""
    self.nowait = nowait

SelectOk

Bases: Frame

Acknowledge confirm mode

This method confirms to the client that the channel was successfully set to use publisher acknowledgements.

Source code in pamqp/commands.py

class SelectOk(base.Frame):
    """Acknowledge confirm mode

    This method confirms to the client that the channel was successfully
    set to use publisher acknowledgements.

    """
    __slots__: typing.ClassVar[list[str]] = []  # AMQ Method Attributes

    frame_id: typing.ClassVar[int] = 11  # AMQP Frame ID
    index: typing.ClassVar[int] = 0x0055000B  # pamqp Mapping Index
    name: typing.ClassVar[str] = 'Confirm.SelectOk'
    synchronous: typing.ClassVar[
        bool] = False  # Indicates if this is a synchronous AMQP method