talus 1.4.0rc1

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 talus

Examples

Creating a consumer which listens on a queue, processes valid messages and publishes as part of processing

Uses default connection parameters and connection retryer expecting a rabbitmq server running in its default configuration.

from talus import DurableConsumer
from talus import DurableProducer
from talus import ConnectionRetryerFactory
from talus import ConsumerConnectionParameterFactory, ProducerConnectionParameterFactory
from talus import MessageProcessorBase
from talus import ConsumeMessageBase, PublishMessageBase, MessageBodyBase
from talus import Queue
from talus import Exchange
from talus import Binding
from typing import Type

##########################
# Consumer Configurations#
##########################
# Configure messages that will be consumed
class ConsumeMessageBody(MessageBodyBase):
    objectName: str
    bucket: str

class ConsumeMessage(ConsumeMessageBase):
    message_body_cls: Type[ConsumeMessageBody] = ConsumeMessageBody

# Configure the queue the messages should be consumed from
inbound_queue = Queue(name="inbound.q")


###########################
# Producer Configurations #
###########################
# Configure messages that will be produced
class ProducerMessageBody(MessageBodyBase):
    key: str
    code: str

class PublishMessage(PublishMessageBase):
    message_body_cls: Type[ProducerMessageBody] = ProducerMessageBody
    default_routing_key: str = "outbound.message.m"

# Configure the queues the message 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 Message -> Outbound Queues)
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.


############################
# Processor Configurations #
############################

# Configure a message processor to handle the consumed messages
class MessageProcessor(MessageProcessorBase):
    def process_message(self, message: ConsumeMessage):
        print(message)
        outbound_message = PublishMessage(
            body=ProducerMessageBody(
                key=message.body.objectName,
                code="newBucket",
            ),
            conversation_id=message.conversation_id,
        )  # crosswalk the values from the consumed message to the produced message
        self.producer.publish(outbound_message)
        print(outbound_message)


# Actually Connect and run the consumer
def main():
    """Starts a listener which will consume messages from the inbound queue and publish messages to the outbound queues."""
    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(message_cls=ConsumeMessage, producer=producer)
            consumer.listen(message_processor)


if __name__ == "__main__":
    # First message to consume
    class InitialMessage(PublishMessageBase):
        message_body_cls: Type[
            ConsumeMessageBody] = ConsumeMessageBody
        default_routing_key: str = "inbound.message.m"

    initial_message_bindings = [Binding(queue=inbound_queue, message=InitialMessage)]

    with DurableProducer(
            queue_bindings=initial_message_bindings,
            publish_exchange=publish_exchange,
            connection_parameters=ProducerConnectionParameterFactory(),
            connection_retryer=ConnectionRetryerFactory(),
    ) as producer:
        producer.publish(InitialMessage(body={"objectName": "object", "bucket": "bucket"}))
    # Consume the message and process it
    main()

Correlation IDs

Talus stores a message’s canonical correlation ID in the AMQP BasicProperties.correlation_id property. Publishers also write the same value to the legacy body.conversationId field during the migration period. Pass conversation_id when publishing a follow-up message, and read message.conversation_id when consuming one:

follow_up = PublishMessage(
    body={"key": "value", "code": "updated"},
    conversation_id=message.conversation_id,
)

Consumed messages without an AMQP correlation ID fall back to body.conversationId. The legacy body field is marked deprecated, so this fallback emits a DeprecationWarning. This allows services to use the compatibility property while operators track messages that still lack AMQP correlation metadata.