Skip to content

Library simplifying messaging queue integrations in Python. It supports RabbitMQ at this point. The goal is to extend to more Messaging Platforms.

Notifications You must be signed in to change notification settings

getphyllo/py-messaging-library

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

13 Commits
 
 
 
 
 
 
 
 

Repository files navigation

Python Messaging Library

Library simplifying messaging queue integrations in Python. It supports RabbitMQ at this point which is a wrapper on top of Pika. The goal is to extend to more Messaging Platforms.

Installation

If you are using pip then
pip install git+https://github.com/getphyllo/py-messaging-library.git@<version_tag>
Ex: pip install git+https://github.com/getphyllo/[email protected]
More information here

If you are using pipenv
pipenv install git+ssh://[email protected]/getphyllo/py-messaging-library@{version_tag}#egg=py_messaging_library
pipenv install git+https://github.com/getphyllo/py-messaging-library.git@{version_tag}#egg=py_messaging_library
Ex: pipenv install git+ssh://[email protected]/getphyllo/[email protected]#egg=py_messaging_library
pipenv install git+https://github.com/getphyllo/[email protected]#egg=py_messaging_library
For more information search for Installing from git: here

Usage

Publish and Consume messages - Below is the Sample code for integrating with this library.
config.py
from rabbitmq_client.broker_config import BrokerConfig

broker_config = BrokerConfig(username="user-test",
                             password="user-test",
                             host="localhost",
                             port=5671,
                             vhost="/",
                             ssl_enabled=False)
publish.py
from datetime import datetime
from uuid import UUID

from rabbitmq_client.publisher import publisher
from rabbitmq_client.queue_config import PublishQueueConfig

from config import broker_config


@publisher.register_serializer
def custom_serializer(obj):
    if isinstance(obj, datetime):
        return obj.__str__()
    if isinstance(obj, UUID):
        return str(obj)


def publish_message(message: dict, routing_key: str, exchange_name: str):
    publish_queue_config: PublishQueueConfig = PublishQueueConfig(routing_key=routing_key,
                                                                  exchange=exchange_name,
                                                                  broker_config=broker_config)
    publisher.publish(queue_config=publish_queue_config, payload=message)
handler.py
import logging

from pika import spec
from rabbitmq_client.consumer.base_handler import BaseHandler


class DemoHandler(BaseHandler):
    def handle_message(self, method: spec.Basic.Deliver, properties: spec.BasicProperties, message: dict):
        print(f'Processing message {message}')
        # do some work
        logging.info('Message processed successfully')
consumer.py
from rabbitmq_client.consumer.multi_queue_listener import MultiQueueListener
from rabbitmq_client.queue_config import ListenQueueConfig

from broker_config import MQ_BROKER
from handler import DemoHandler

queue1 = ListenQueueConfig(name="queue_name",
                          handler=DemoHandler(),
                          broker_config=MQ_BROKER)

mq_listener = MultiQueueListener([queue1])
mq_listener.listen()
demo.py
import logging

from rabbitmq_client.consumer.multi_queue_listener import MultiQueueListener
from rabbitmq_client.queue_config import ListenQueueConfig

from config import broker_config
from handler import DemoHandler
from publish import publish_message


if __name__ == '__main__':
    # Sending message to exchange with a routing key
    message = {'key': 'value'}
    logging.info('Sending message')
    publish_message(message=message, exchange_name='demo-exchange', routing_key='.queue-1.queue-2.')
    logging.info('Message sent')

    # Processing messages from queue
    queue1 = ListenQueueConfig(name='demo-queue-1',
                               handler=DemoHandler(),
                               broker_config=broker_config)
    queue2 = ListenQueueConfig(name='demo-queue-2',
                               handler=DemoHandler(),
                               broker_config=broker_config)

    multi_queue_listener = MultiQueueListener([queue1, queue2])
    multi_queue_listener.listen()

When you run demo.py the script pushes message to Rabbit MQ, and also starts a consumer on those queues

About

Library simplifying messaging queue integrations in Python. It supports RabbitMQ at this point. The goal is to extend to more Messaging Platforms.

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Languages