Hide navigation sidebar
Hide table of contents sidebar
Back to top
Toggle table of contents sidebar

# Source code for aio_pika.exceptions

    import asyncio

    import pamqp.exceptions
    from aiormq.exceptions import (
        AMQPChannelError, AMQPConnectionError, AMQPError, AMQPException,
        AuthenticationError, ChannelClosed, ChannelInvalidStateError,
        ChannelNotFoundEntity, ChannelPreconditionFailed, ConnectionClosed,
        DeliveryError, DuplicateConsumerTag, IncompatibleProtocolError,
        InvalidFrameError, MethodNotImplemented, ProbableAuthenticationError,
        ProtocolSyntaxError, PublishError,
    )

    CONNECTION_EXCEPTIONS = (
        AMQPError,
        ConnectionError,
        OSError,
        RuntimeError,
        StopAsyncIteration,
        pamqp.exceptions.PAMQPException,
    )

    class MessageProcessError(AMQPError):
        reason = "%s: %r"

    class QueueEmpty(AMQPError, asyncio.QueueEmpty):
        pass

    __all__ = (
        "AMQPChannelError",
        "AMQPConnectionError",
        "AMQPError",
        "AMQPException",
        "AuthenticationError",
        "CONNECTION_EXCEPTIONS",
        "ChannelClosed",
        "ChannelInvalidStateError",
        "ChannelNotFoundEntity",
        "ChannelPreconditionFailed",
        "ConnectionClosed",
        "DeliveryError",
        "DuplicateConsumerTag",
        "IncompatibleProtocolError",
        "InvalidFrameError",
        "MessageProcessError",
        "MethodNotImplemented",
        "ProbableAuthenticationError",
        "ProtocolSyntaxError",
        "PublishError",
        "QueueEmpty",
    )