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",
)