Integrating Robyn with RabbitMQ | Python

In this article, we are going to learn how to integrate RabbitMQ into a Robyn server.

This article will show the code of the server that consumes messages from a RabbitMQ queue. And the code of a server that publishes messages to a queue.

RabbitMQ

RabbitMQ is a messaging broker - an intermediary for messaging. It gives your applications a common platform to send and receive messages, and your messages a safe place to live until received.

Requirements

• Python installed

• Pip installed

• RabbitMQ installed(Download it from here)

Consumer

In this section, we are going to build a server that consumes messages from a queue.

First, we have to build a program that sends messages to the queue. So, that is what sender.py does. And in the app.py file, we write the server that consumes the messages.

sender.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')

channel.basic_publish(exchange='', routing_key='hello', body="Hello World!")
print(" [x] Sent 'Hello'")
connection.close()

In the sender.py file, we import the RabbitMQ library, pika. We create a connection to the RabbitMQ server that runs on localhost. Then, we open a channel. Channels are where messages are routed and queued.

We declare a queue named "hello", where messages will be published and consumed from. We publish the "Hello World!" message to the "hello" queue. When the program publishes the message, it will the message " [x] Sent 'Hello' " in the console. Then we close the connection with the RabbitMQ server.

We run this program and we should see the following message in our terminal.

app.py

from robyn import Robyn, status_codes

from robyn.robyn import Response
import asyncio
import aio_pika

app = Robyn( __file__ )

@app.get("/")
async def hello():
    return Response(status_code=status_codes.HTTP_200_OK, headers={}, body="Hello, World!")

async def startup():
    loop = asyncio.get_event_loop()
    connection = await aio_pika.connect("amqp://guest:guest@localhost/", loop = loop)
    channel = await connection.channel()
    queue = await channel.declare_queue("hello")

    def callback(message: aio_pika.IncomingMessage):
        body = message.body.decode("utf-8")

        print(body)
    await queue.consume(callback)    

app.startup_handler(startup)

app.start(port=8000, url="0.0.0.0")

In the app.py file, we import the aio-pika library which is an async library for RabbitMQ. Then, we create a Robyn server, with one endpoint that sends a "Hello World" as a response. We define the startup() function that receives messages from the RabbitMQ server.

In the startup() function, we create a connection to the RabbitMQ server, create a channel, and declare the queue it will consume messages. Then, we define the callback() function, it receives the messages and prints it.

We use startup_handler() to register the startup() function, this way the server will start consuming messages from the RabbitMQ queue when the API starts.

We start the server and we should receive the message from the queue printed in the terminal.

Publisher

Here, we are going to build a server that publishes messages to a queue and a program that consumes the messages.

The receiver.py file is where we write the program that will consume messages from the queue. And app.py file is where we write the server that publishes them.

receiver.py

import pika, sys, os

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()

    channel.queue_declare(queue='hello')

    def callback(ch, method, properties, body):
        print(f" [x] Received {body}")

    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == ' __main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

Here we create a connection to the RabbitMQ server running locally and subscribe to the 'hello' queue. When messages are received from that queue, the callback function is called which prints the message body.

app.py

Here, we are going to create the server to publish messages to the "hello" queue.

from robyn import Robyn, status_codes

from robyn.robyn import Response
import asyncio
import pika

app = Robyn( __file__ )

@app.get("/")
async def hello():
    body = "Hello World!"
    await sender(body)
    return Response(status_code=status_codes.HTTP_200_OK, headers={}, body="Hello, World!")

async def sender(body: str):
    loop = asyncio.get_event_loop()
    connection = await aio_pika.connect("amqp://guest:guest@localhost/", loop = loop)
    channel = await connection.channel()

    await channel.default_exchange.publish(aio_pika.Message(body=body.encode()), routing_key='hello')
    print(" [x] Sent 'Hello'")
    await connection.close()

app.start(port=8000, url="0.0.0.0")

We create the sender() function that creates a connection to the RabbitMQ server, then we create a channel and declare the queue where we want to publish the messages.

Then, we send to the queue the body that the sender() function receives.

We start the server and navigate or make a request to localhost:8000/. We should see the "[x] Sent 'Hello' " message in the terminal.

In the terminal of the receiver.py file we run the program, we should see the following message:

Conclusion

In conclusion, using RabbitMQ as a message broker, you can decouple your different components and scale them independently. This way we can build Robyn microservices that communicate with each other.

If you have any recommendations about other packages, architectures, how to improve my code, my English, or anything; please leave a comment or contact me through Twitter, or LinkedIn.

The source code is here.

Resources

RabbitMQ Tutorial One

aio-pika library documentation

Robyn documentation