Minimal working example for Nicegui and Rabbitmq

[Twansklaf]

Hi community,
I couldn't find examples of consuming messages from RabbitMQ and using UI components from nicegui, so I pulled my hair a little bit building this example as I find the integration of both those packages asyncio non trivial. Many other methods I have tried were blocking one of the two event loops, despite trying multiple methods documented in nicegui's reference or GH issues.

I usually use pika to interact with RabbitMQ instances, but in this case there is no way I could make it work. I then picked aiormq that provides a fully asyncio-based RMQ implementation, with a syntax that is very close to pika's.

This simple example runs with docker compose.

compose.yaml

services:
  main_chat:
    depends_on:
      queue_chat:
        condition: service_healthy
    image: python_xxx
    ports:
      - "8080:8080"
    environment:
      PYTHONPATH: /mnt/src/
    command: python your_path/nicegui_test_rmq_consumer.py
    volumes:
      - 'your_project_path:/mnt'
    networks:
      chattest:
        ipv4_address: 170.0.0.3

  data_producer:
    depends_on:
      queue_chat:
        condition: service_healthy
    image: python_xxx
    environment:
      PYTHONPATH: /mnt/src
    command: python your_path/nicegui_tes_producer.py
    volumes:
      - 'your_project_path:/mnt'
    networks:
      chattest:
        ipv4_address: 170.0.0.4

  queue_chat:
    image: rabbitmq:latest
    networks:
      chattest:
        ipv4_address: 170.0.0.2
    logging:
      driver: none
    healthcheck:
      test: rabbitmq-diagnostics -q ping
      interval: 500ms
      timeout: 30s
      retries: 100

networks:
  chattest:
    driver: bridge
    ipam:
      config:
        - subnet: 170.0.0.0/8
          gateway: 170.0.0.1

producer.py

import pika
import time
import uuid
import json
import datetime
from pika.exceptions import AMQPConnectionError, AMQPChannelError

RMQ_IP="fill_in"
RMQ_PORT="fill_in"

connected = False
while not connected :
	try :
		connection = pika.BlockingConnection(
			pika.ConnectionParameters(RMQ_IP, RMQ_PORT))

	except (AMQPConnectionError, AMQPChannelError) :
		time.sleep(.5)

	else :
		channel = connection.channel()
		connected = True

channel.queue_declare("chat_comm")
counter = 0

while True :
	uid = str(uuid.uuid4())
	message = {
		"text" : f"This is the message no {counter}",
		"user_id": uid,
		"stamp": str(datetime.datetime.now()),
		"avatar" : f"https://robothash.org/{uid}?bgset=bg3"
	}

	channel.basic_publish(exchange="", routing_key="chat_comm",
		body = json.dumps(message))
	counter += 1
	time.sleep(2)

consumer.py

import json
import time
import aiormq
import datetime
import asyncio
from uuid import uuid4
from nicegui import ui, run, app
from aiormq.exceptions import AMQPConnectionError

RMQ_IP="fill_in"
RMQ_PORT="fill_in"

own_id = str(uuid4())
global_messages = []

async def read_queue() -> None :
	print("Hard wait for connection", flush=True)
	
	# unfortunately necessary as I didn't find any simple and straightforward way
	# to combine aiormq, nicegui's asyncio loop and execption handling for 
	# AMQPConnection, in the case of a docker compose where the rabbitmq is slower
	# to start for instance
	time.sleep(3)

	connection = await aiormq.connect(f"http://{RMQ_IP}:{RMQ_PORT}")
	channel = await connection.channel()
	declare_ok = await channel.queue_declare("chat_comm")

	print(f"Connection statuses : {connection}{channel}{declare_ok}",
		flush=True)
	
	while not app.is_stopped or (queue_state.method.message_count == 0):
		msg = await channel.basic_get("chat_comm")
		if msg.body.decode() != "":

			try :
				global_messages.append(json.loads(msg.body.decode()))

			except json.JSONDecodeError :
				print(f"FAILED JSON CONTENT: {msg}", flush=True)

			else :
				chat_messages.refresh()

		else :
			pass

		await asyncio.sleep(3)

@ui.refreshable
def chat_messages(opt_msg_list=[]) -> None :
	for msg_obj in global_messages + opt_msg_list:
		ui.chat_message(
			text=msg_obj["text"],
			stamp=msg_obj["stamp"],
			avatar=msg_obj["avatar"],
			sent=(own_id == msg_obj["user_id"])
		)
	ui.run_javascript('window.scrollTo(0, document.body.scrollHeight)')

@ui.page('/')
async def main() :
	"""Similar to text chat example from Nicegui's documentation
	"""
	avatar = f"https://robohash.org/{own_id}?bgset=set3"

	ui.add_css(r'a:link, a:visited {color: inherit !important; text-decoration: none; font-weight: 500}')
	with ui.footer().classes('bg-white'), ui.column().classes('w-full max-w-3xl mx-auto my-6'):
		with ui.row().classes('w-full no-wrap items-center'):
			with ui.avatar().on('click', lambda: ui.navigate.to(main)):
				ui.image(avatar)
		ui.markdown('DecisionGPT') \
			.classes('text-xs self-end mr-8 m-[-1em] text-primary')

	await ui.context.client.connected()  # chat_messages(...) uses run_javascript which is only possible after connecting
	ui.label("No messages yet").classes("mx-auto my-36")
	chat_messages()

if __name__ in ["__main__", "__mp_main__"] :
	ui.run()
	app.on_startup(read_queue)