Radicale/radicale/hook/rabbitmq/__init__.py

import pika

from radicale import hook
from radicale.hook import HookNotificationItem


class Hook(hook.BaseHook):

    def __init__(self, configuration):
        super().__init__(configuration)
        endpoint = configuration.get("hook", "rabbitmq_endpoint")
        self._topic = configuration.get("hook", "rabbitmq_topic")
        self._encoding = configuration.get("encoding", "stock")

        self._make_connection_synced(endpoint)
        self._make_declare_queue_synced(self._topic)

    def _make_connection_synced(self, endpoint):
        parameters = pika.URLParameters(endpoint)
        connection = pika.BlockingConnection(parameters)
        self._channel = connection.channel()

    def _make_declare_queue_synced(self, topic):
        self._channel.queue_declare(queue=topic, durable=True)

    def notify(self, notification_item):
        if isinstance(notification_item, HookNotificationItem):
            self._channel.basic_publish(
                exchange='',
                routing_key=self._topic,
                body=notification_item.to_json().encode(
                    encoding=self._encoding
                )
            )