carehare 1.0.2

Installation

pip install carehare

Usage

Consumer (async for):

try:
    async with carehare.connect("amqps://guest:guest@localhost") as connection:
        try:
            await connection.queue_declare("my-queue", exclusive=True)
            async with connection.acking_consumer("my-queue") as consumer:
                async for message in consumer:
                    print(repr(message))
                    # ... if we raise an exception here, we won't ack.
                    #
                    # ... if we `break` from this loop, we won't ack.
                    #
                    # Call `consumer.close()` before (or instead of) `break`
                    # to abort iteration.
        except carehare.ChannelClosedByServer:
            logger.info("RabbitMQ told this one consumer to go away")
        except carehare.ConnectionClosed:
            # Either RabbitMQ is telling us an error (and the outer context
            # manager will throw it), or we called connection.close()
            # ourselves (so we want to close).
            pass
except carehare.ConnectionClosedByServer:
    # str(error) will give the RabbitMQ error message
    logger.error("RabbitMQ closed our connection")
except carehare.ConnectionClosedByHeartbeatMonitor:
    logger.error("RabbitMQ went away")

Consumer (next_delivery):

try:
    async with carehare.connect("amqps://guest:guest@localhost") as connection:
        try:
            await connection.queue_declare("my-queue", exclusive=True)
            async with connection.acking_consumer("my-queue") as consumer:
                while True:
                    message, delivery_tag = await consumer.next_delivery()
                    # You must ack() (with no await). If RabbitMQ doesn't
                    # receive this ack, it may deliver the same message to
                    # another client.
                    consumer.ack(delivery_tag)
                    if message.startswith(b"okay, go away now"):
                        break
        except carehare.ChannelClosedByServer:
            logger.info("RabbitMQ told this one consumer to go away")
        except carehare.ConnectionClosed:
            # Either RabbitMQ is telling us an error (and the outer context
            # manager will throw it), or we called connection.close()
            # ourselves (so we want to close).
            pass
except carehare.ConnectionClosedByServer:
    # str(error) will give the RabbitMQ error message
    logger.error("RabbitMQ closed our connection")
except carehare.ConnectionClosedByHeartbeatMonitor:
    logger.error("RabbitMQ went away")

Publisher:

try:
    async with carehare.connect("amqps://guest:guest@localhost") as connection:
        try:
            await connection.publish(b"Hello, world!", routing_key="my-queue")
        except carehare.ServerSentNack:
            logger.warn("Failed to publish message")
        except carehare.ChannelClosedByServer:
            # str(err) will give the RabbitMQ error message -- for instance,
            # "404 NOT_FOUND" if the exchange does not exist
            logger.error("Problem with the exchange")
except carehare.ConnectionClosedByServer:
    # str(error) will give the RabbitMQ error message
    logger.error("RabbitMQ closed our connection")
except carehare.ConnectionClosedByHeartbeatMonitor:
    logger.error("RabbitMQ went away")

Design decisions

carehare is designed to turn RabbitMQ’s asynchronous error system into _understandable_ Python exceptions.

Dependencies

You’ll need Python 3.8+ and a RabbitMQ server.

If you have Docker, here’s how to start a development server:

test-server/prepare-certs.sh  # Create SSL certificates used in tests
docker run --rm -it \
     -p 5671:5671 \
     -p 5672:5672 \
     -p 15672:15672 \
     -v "/$(pwd)"/test-server:/test-server \
     -e RABBITMQ_SSL_CACERTFILE=/test-server/ca.cert \
     -e RABBITMQ_SSL_CERTFILE=/test-server/server.cert \
     -e RABBITMQ_SSL_KEYFILE=/test-server/server.key \
     -e RABBITMQ_SSL_VERIFY=verify_peer \
     -e RABBITMQ_SSL_FAIL_IF_NO_PEER_CERT=true \
     -e RABBITMQ_CONFIG_FILE=/test-server/rabbitmq \
     rabbitmq:3.8.11-management-alpine

During testing, see the RabbitMQ management interface at http://localhost:15672.

Contributing