Python Rabbit wrapper to simplify to use Exchanges and Queues with decorators
import os

BROKER_PORT = "BROKER_PORT"
BROKER_PWD = "BROKER_PWD"
BROKER_USERNAME = "BROKER_USERNAME"
BROKER_SERVER = "BROKER_SERVER"


def get_broker_url():
    server = os.environ[BROKER_SERVER] if BROKER_SERVER in os.environ else None
    username = os.environ[BROKER_USERNAME] if BROKER_USERNAME in os.environ else None
    pwd = os.environ[BROKER_PWD] if BROKER_PWD in os.environ else None
    port = os.environ[BROKER_PORT] if BROKER_PORT in os.environ else "5672"
    return f'amqp://{username}:{pwd}@{server}:{port}' if server else 'amqp://rabbitmq:rabbitmq@localhost:5672'

def init_os_environ(host, username, password, port):
    os.environ[BROKER_SERVER] = host
    os.environ[BROKER_USERNAME] = username
    os.environ[BROKER_PWD] = password
    os.environ[BROKER_PORT] = str(port)

hijiki/broker/broker_data.py
asengardeon/hijiki
import os

from celery import Celery
from kombu import Exchange, Queue

from hijiki.broker.broker_data import get_broker_url, init_os_environ


class HijikiBroker:
    celery_broker = None

    def bind_queues(self, queues_names):
        for name in queues_names:
            dlx = Exchange(f'dlq_{name}_exchange', type='topic')

            dead_letter_queue = Queue(f'dlq_{name}', dlx, routing_key='*', queue_arguments={'x-queue-type': 'quorum'})
            dead_letter_queue.bind(self.celery_broker.broker_connection()).declare()

    def __init__(self, app_name, host, username, password, port):
        init_os_environ(host, username, password, port)
        self.celery_broker = Celery(app_name, broker=get_broker_url(), set_as_current=True)


    def get_celery_broker(self):
        return self.celery_broker

    def bind_queues(self, queues_names):
        self.bind_queues(queues_names)
hijiki/broker/hijiki_broker.py
asengardeon/hijiki
from functools import wraps, partial
from typing import List

from kombu import Queue, Exchange
from kombu.common import logger

from hijiki.broker.hijiki_broker import HijikiBroker
from hijiki.decorator.worker import Worker

class HijikiQueueExchange():
    def __init__(self, name, exchange_name):
        self.name = name
        self.exchange_name = exchange_name

class HijikiRabbit():

    queues = {}
    callbacks = {}
    _prefix = ""
    def with_queues_exchange(self, queue_exchanges: List[HijikiQueueExchange]):
        self.queue_exchanges = queue_exchanges
        return self

    def with_username(self, username: str):
        self.username = username
        return self

    def with_password(self, password: str):
        self.password = password
        return self
    def with_host(self, host: str):
        self.host = host
        return self

    def with_port(self, port: str):
        self.port = port
        return self

    def build(self):
        self.broker = HijikiBroker('worker', self.host, self.username, self.password, self.port)
        self.connection = self.broker.get_celery_broker().broker_connection()
        self.init_queues()
        return self

    def init_queues(self, ):
        for q in self.queue_exchanges:
            name = q.name
            if name not in self.queues:
                self.queues[name] = []
            if name not in self.callbacks:
                self.callbacks[name] = []

            logger.debug("Setting up %s" % name)
            routing_key = "*"

            task_exchange = Exchange(f'{q.exchange_name}', type='topic')
            task_exchange_dlq = Exchange(f'{q.exchange_name}_dlq', type='topic')

            queue = Queue(name,
                          task_exchange,
                          routing_key=routing_key,
                          queue_arguments={'x-queue-type': 'quorum','x-dead-letter-exchange': f'{q.exchange_name}_dlq', 'x-delivery-limit': 10})

            queue.bind(self.connection).declare()

            queue_dlq = Queue(f'{name}_dlq',
                          task_exchange_dlq,
                          routing_key=routing_key,
                          queue_arguments={'x-queue-type': 'quorum'})

            queue_dlq.bind(self.connection).declare()

            self.queues[name].append(queue)


    def _wrap_function(self, function, callback, queue_name, task=False):

        self.callbacks[queue_name].append(callback)

        # The function returned by the decorator don't really do
        # anything.  The process_msg callback added to the consumer
        # is what actually responds to messages  from the client
        # on this particular queue.

        def decorate(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                pass
            return wrapper
        return decorate

    def task(self, func=None, *, queue_name=None):
        """Wrap around a function that should be a task.
        The client should not expect anything to be returned.

        """
        if func is None:
            return partial(self.task, queue_name=queue_name)

        def process_message(body, message):
            logger.debug("Processing function {!r} "
                         " in message: {!r} "
                         "with data {!r}".format(func.__name__,
                                                 message,
                                                 body))
            try:
                func(body)
            except Exception:
                logger.error("Problem processing task", exc_info=True)
            else:
                logger.debug("Ack'ing message.")
                message.ack()

        return self._wrap_function(
            func, process_message, queue_name, task=True)


    def run(self):
        try:
            worker = Worker(self.connection, self)
            worker.run()
        except KeyboardInterrupt:
            print('bye bye')
hijiki/broker/hijiki_rabbit.py
asengardeon/hijiki
from kombu import producers, Exchange

from hijiki.broker.hijiki_broker import HijikiBroker


class Publisher():

    def __init__(self, host, username, password, port):
        self.client = HijikiBroker('client', host, username, password, port)

    def publish_message(self, event_name: str, data: str):
        payload = {"value": data}
        connection = self.client.get_celery_broker().broker_connection()
        with producers[connection].acquire(block=True) as producer:
            task_exchange = Exchange(event_name, type='topic')
            producer.publish(payload,  exchange=task_exchange, declare=[task_exchange], routing_key='x')

hijiki/publisher/Publisher.py
asengardeon/hijiki
from setuptools import setup, find_packages

setup(
    name="hijiki",
    version="{{VERSION_PLACEHOLDER}}",
    description="Python Rabbit wrapper library to simplify to use Exchanges and Queues with decorators",
    author="Leandro Vilson Battisti",
    keywords=['Celery', 'Kombu', 'RabbitMQ', 'decorator'],
    packages=find_packages(),
    long_description=open('README.md').read(),
    long_description_content_type='text/markdown',
)
setup.py
asengardeon/hijiki
import threading
import time
import unittest

from testcontainers.core.container import DockerContainer

from hijiki.broker.hijiki_rabbit import HijikiQueueExchange, HijikiRabbit
from hijiki.publisher.Publisher import Publisher

result_event_list = []
container = (DockerContainer("rabbitmq:3-management-alpine") \
    .with_bind_ports(15672, 15672) \
    .with_bind_ports(5672, 5672) \
    .with_name("rabbit_test_container") \
    .with_env("RABBITMQ_DEFAULT_USER", "rabbitmq") \
    .with_env("RABBITMQ_DEFAULT_PASS", "rabbitmq").start())
class Runner():

    qs = [HijikiQueueExchange('teste1', 'teste1_event'), HijikiQueueExchange('teste2', 'teste2_event')]
    gr = HijikiRabbit().with_queues_exchange(qs) \
        .with_username("rabbitmq") \
        .with_password("rabbitmq") \
        .with_host(container.get_container_host_ip()) \
        .with_port(5672) \
        .build()
    @gr.task(queue_name="teste1")
    def internal_consumer(self):
        print("consumiu")
        result_event_list.append('recebeu evento')

    def run(self):
        threads = []
        t = threading.Thread(target=self.gr.run)
        threads.append(t)
        t.start()

    def __del__(self):
        self.gr.close()
        super()


class TestPublisherConsumer(unittest.TestCase):
    runner = None
    def setUp(self):
        if not self.runner:
            self.runner = Runner()
            self.runner.run()

    def test_publish_a_message(self):
        pub = Publisher("localhost", "rabbitmq", "rabbitmq", 5672)
        pub.publish_message('teste1_event', '{"value": "Esta é a mensagem"}')


    def test_consume_a_message(self):
        pub = Publisher("localhost", "rabbitmq", "rabbitmq", 5672)
        time.sleep(5)
        pub.publish_message('teste1_event', '{"value": "Esta é a mensagem"}')
        time.sleep(1)
        self.assertEqual(len(result_event_list), 1)


tests/publisher_consumer_test.py
asengardeon/hijiki
import os
import unittest

from hijiki.broker.broker_data import get_broker_url, BROKER_SERVER, BROKER_USERNAME, BROKER_PWD, BROKER_PORT

class TestMyModule(unittest.TestCase):

    def setUp(self):
        os.environ[BROKER_SERVER] = 'teste'
    def tearDown(self):
        os.environ.pop(BROKER_SERVER) if os.environ.get(BROKER_SERVER) else None
        os.environ.pop(BROKER_USERNAME) if os.environ.get(BROKER_USERNAME) else None
        os.environ.pop(BROKER_PWD) if os.environ.get(BROKER_PWD) else None
        os.environ.pop(BROKER_PORT) if os.environ.get(BROKER_PORT) else None

    def test_broker_username_env_not_exists_in_environment_variable_but_server_exists(self):
        self.assertEqual(get_broker_url(), 'amqp://None:None@teste:5672')

    def test_broker_pwd_env_not_exists_in_environment_variable_but_server_exists(self):
        self.assertEqual(get_broker_url(), 'amqp://None:None@teste:5672')

    def test_broker_port_env_not_exists_in_environment_variable_but_server_exists(self):
        self.assertEqual(get_broker_url(), 'amqp://None:None@teste:5672')

    def test_broker_server_env_exists_in_environment_variable(self):
        self.assertEqual(get_broker_url(), 'amqp://None:None@teste:5672')

    def test_broker_server_env_exists_in_environment_variable(self):
        self.assertEqual(get_broker_url(), 'amqp://None:None@teste:5672')

    def test_alldata_env_exists_in_environment_variable(self):
        os.environ[BROKER_SERVER] = 'server'
        os.environ[BROKER_USERNAME] = 'usr'
        os.environ[BROKER_PWD] = 'password'
        os.environ[BROKER_PORT] = '5427'
        self.assertEqual(get_broker_url(), 'amqp://usr:password@server:5427')

tests/test_broker_data_server_exists.py
asengardeon/hijiki