A wrapper for connecting to RabbitMQ which constrains clients to a single purpose channel (producer or consumer) with healing for intermittent connectivity.
Project description
talus (noun) - ta·lus | ˈtā-ləs: a slope formed especially by an accumulation of rock debris; Occasional habitat of the pika.
A wrapper for connecting to RabbitMQ which constrains clients to a single purpose channel (producer or consumer) with healing for intermittent connectivity.
Features
Guided separation of connections for producers and consumers
Re-establish connections to the server when lost
Constrained interface to support simple produce / consume use cases for direct exchanges
Installation
pip install talusExamples
Creating a consumer which listens on a queue, processes valid messages and publishes as part of processing
from talus.consumer DurableConsumer
from talus.producer DurableProducer
from talus.models.retryer import ConnectionRetryerFactory
from talus.models.connection_parameters import ConsumerConnectionParameterFactory, ProducerConnectionParameterFactory
from talus.models.processor import MessageProcessorBase
from talus.models.message import ConsumeMessageBase, PublishMessageBase, MessageBodyBase
from talus.models.queue import Queue
from talus.models.exchange import Exchange
from talus.models.binding import Binding
from typing import Type
############
# Consumer #
############
# Configure messages that will be consumed
class ObjectMessageBody(MessageBodyBase):
objectName: str
bucket: str
class ConsumeMessage(ConsumeMessageBase):
message_body_cls: Type[ObjectMessageBody] = ObjectMessageBody
# Configure the queue the messages should be consumed from
inbound_queue = Queue(name="inbound.q")
# Configure a message processor to handle the consumed messages
class MessageProcessor(MessageProcessorBase):
def process_message(self, message: ConsumeMessage):
print(f"Received message: {message}")
outbound_message = PublishMessage(
body=ObjectMessageBody(objectName=message.body.objectName, bucket="newBucket"),
) # change the bucket name for some reason
self.producer.publish(outbound_message)
############
# Producer #
############
# Configure messages that will be produced
class PublishMessage(PublishMessageBase):
message_body_cls: Type[ObjectMessageBody] = ObjectMessageBody # using the same schema for simplicity
default_routing_key: str = "outbound.message.m"
# Configure the queue the messages should be routed to
outbound_queue_one = Queue(name="outbound.one.q")
outbound_queue_two = Queue(name="outbound.two.q")
# Configure the exchange and queue bindings for publishing
publish_exchange = Exchange(name="outbound.exchange") # Direct exchange by default
bindings = [Binding(queue=outbound_queue_one, message=PublishMessage),
Binding(queue=outbound_queue_two, message=PublishMessage)] # publishing PublishMessage will route to both queues.
# Actually Connect and run the consumer
def main():
with DurableProducer(
queue_bindings=bindings,
publish_exchange=publish_exchange,
connection_parameters=ProducerConnectionParameterFactory(),
connection_retryer=ConnectionRetryerFactory(),
) as producer:
with DurableConsumer(
consume_queue=inbound_queue,
connection_parameters=ConsumerConnectionParameterFactory(),
connection_retryer=ConnectionRetryerFactory(),
) as consumer:
message_processor = MessageProcessor(producer=producer)
consumer.listen(message_processor)
if __name__ == "__main__":
main()Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
talus-1.0.0rc7.tar.gz
(21.5 kB
view details)
Built Distribution
talus-1.0.0rc7-py3-none-any.whl
(25.2 kB
view details)
File details
Details for the file talus-1.0.0rc7.tar.gz.
File metadata
- Download URL: talus-1.0.0rc7.tar.gz
- Upload date:
- Size: 21.5 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.0 CPython/3.11.9
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 696e5f36d308b4b10f06c295137bc6f6122fdc5346a20dde7c0696dffe29d1c7 |
|
| MD5 | 9a9f161df6786b9a1f46413b3094b934 |
|
| BLAKE2b-256 | 2eced53d5b56a84c3c958468e72c89c9c6a3755bdcffbdae95771b13e847645e |
Provenance
File details
Details for the file talus-1.0.0rc7-py3-none-any.whl.
File metadata
- Download URL: talus-1.0.0rc7-py3-none-any.whl
- Upload date:
- Size: 25.2 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/5.1.0 CPython/3.11.9
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | 7613117a2bf9d428ae88cb46d18caf4d3acc95911f1160c29e392d3adf2ca9c6 |
|
| MD5 | 7b2075dd0ef92c3a757f459a8888b519 |
|
| BLAKE2b-256 | a4b77ccaddd70d549f0afc589557da65e7f0d7ce67f4037535d9ae7d5e213f6f |
