Skip to content

rabbitmq#

RabbitMQ Message Queue.

RabbitMQMessageQueueConfig #

Bases: BaseSettings

RabbitMQ message queue configuration.

Parameters:

Name Type Description Default
type Literal[str]
'rabbitmq'
url str
'amqp://guest:guest@localhost/'
exchange_name str
'llama-deploy'
username str | None
None
password str | None
None
host str | None
None
port int | None
None
vhost str | None
None
secure bool | None
None
Source code in llama_deploy/message_queues/rabbitmq.py
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class RabbitMQMessageQueueConfig(BaseSettings):
    """RabbitMQ message queue configuration."""

    model_config = SettingsConfigDict(env_prefix="RABBITMQ_")

    type: Literal["rabbitmq"] = Field(default="rabbitmq", exclude=True)
    url: str = DEFAULT_URL
    exchange_name: str = DEFAULT_EXCHANGE_NAME
    username: Optional[str] = None
    password: Optional[str] = None
    host: Optional[str] = None
    port: Optional[int] = None
    vhost: Optional[str] = None
    secure: Optional[bool] = None

    def model_post_init(self, __context: Any) -> None:
        if self.username and self.password and self.host:
            scheme = "amqps" if self.secure else "amqp"
            self.url = f"{scheme}://{self.username}:{self.password}@{self.host}"
            if self.port:
                self.url += f":{self.port}"
            elif self.vhost:
                self.url += f"/{self.vhost}"

RabbitMQMessageQueue #

Bases: BaseMessageQueue

RabbitMQ integration with aio-pika client.

This class creates a Work (or Task) Queue. For more information on Work Queues with RabbitMQ see the pages linked below: 1. https://aio-pika.readthedocs.io/en/latest/rabbitmq-tutorial/2-work-queues.html 2. https://aio-pika.readthedocs.io/en/latest/rabbitmq-tutorial/3-publish-subscribe.html

Connections are established by url that use amqp uri scheme (https://www.rabbitmq.com/docs/uri-spec#the-amqp-uri-scheme) amqp_URI = "amqp://" amqp_authority [ "/" vhost ][ "?" query ] amqp_authority = [ amqp_userinfo "@" ] host [ ":" port ] amqp_userinfo = username [ ":" password ] username = ( unreserved / pct-encoded / sub-delims ) password = ( unreserved / pct-encoded / sub-delims ) vhost = segment

Parameters:

Name Type Description Default
url str
'amqp://guest:guest@localhost/'
exchange_name str
'llama-deploy'
The Work Queue created has the following properties
  • Exchange with name self.exchange
  • Messages are published to this queue through the exchange
  • Consumers are bound to the exchange and have queues based on their message type
  • Round-robin dispatching: with multiple consumers listening to the same queue, only one consumer will be chosen dictated by sequence.

Attributes:

Name Type Description
url str

The amqp url string to connect to the RabbitMQ server

exchange_name str

The name to give to the so-called exchange within RabbitMQ AMQP 0-9 protocol.

Examples:

from llama_deploy.message_queues.rabbitmq import RabbitMQMessageQueue

message_queue = RabbitMQMessageQueue()  # uses the default url
Source code in llama_deploy/message_queues/rabbitmq.py
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
class RabbitMQMessageQueue(BaseMessageQueue):
    """RabbitMQ integration with aio-pika client.

    This class creates a Work (or Task) Queue. For more information on Work Queues
    with RabbitMQ see the pages linked below:
        1. https://aio-pika.readthedocs.io/en/latest/rabbitmq-tutorial/2-work-queues.html
        2. https://aio-pika.readthedocs.io/en/latest/rabbitmq-tutorial/3-publish-subscribe.html

    Connections are established by url that use amqp uri scheme
    (https://www.rabbitmq.com/docs/uri-spec#the-amqp-uri-scheme)
        amqp_URI       = "amqp://" amqp_authority [ "/" vhost ] [ "?" query ]
        amqp_authority = [ amqp_userinfo "@" ] host [ ":" port ]
        amqp_userinfo  = username [ ":" password ]
        username       = *( unreserved / pct-encoded / sub-delims )
        password       = *( unreserved / pct-encoded / sub-delims )
        vhost          = segment

    The Work Queue created has the following properties:
        - Exchange with name self.exchange
        - Messages are published to this queue through the exchange
        - Consumers are bound to the exchange and have queues based on their
            message type
        - Round-robin dispatching: with multiple consumers listening to the same
            queue, only one consumer will be chosen dictated by sequence.

    Attributes:
        url (str): The amqp url string to connect to the RabbitMQ server
        exchange_name (str): The name to give to the so-called exchange within
            RabbitMQ AMQP 0-9 protocol.

    Examples:
        ```python
        from llama_deploy.message_queues.rabbitmq import RabbitMQMessageQueue

        message_queue = RabbitMQMessageQueue()  # uses the default url
        ```
    """

    url: str = DEFAULT_URL
    exchange_name: str = DEFAULT_EXCHANGE_NAME

    def __init__(
        self,
        url: str = DEFAULT_URL,
        exchange_name: str = DEFAULT_EXCHANGE_NAME,
        **kwargs: Any,
    ) -> None:
        super().__init__(url=url, exchange_name=exchange_name)

    @classmethod
    def from_url_params(
        cls,
        username: str,
        password: str,
        host: str,
        vhost: str = "",
        port: Optional[int] = None,
        secure: bool = False,
        exchange_name: str = DEFAULT_EXCHANGE_NAME,
    ) -> "RabbitMQMessageQueue":
        """Convenience constructor from url params.

        Args:
            username (str): username for the amqp authority
            password (str): password for the amqp authority
            host (str): host for rabbitmq server
            port (Optional[int], optional): port for rabbitmq server. Defaults to None.
            secure (bool, optional): Whether or not to use SSL. Defaults to False.
            exchange_name (str, optional): The exchange name. Defaults to DEFAULT_EXCHANGE_NAME.

        Returns:
            RabbitMQMessageQueue: A RabbitMQ MessageQueue integration.
        """
        if not secure:
            if port:
                url = f"amqp://{username}:{password}@{host}:{port}/{vhost}"
            else:
                url = f"amqp://{username}:{password}@{host}/{vhost}"
        if secure:
            if port:
                url = f"amqps://{username}:{password}@{host}:{port}/{vhost}"
            else:
                url = f"amqps://{username}:{password}@{host}/{vhost}"
        return cls(url=url, exchange_name=exchange_name)

    async def new_connection(self) -> "Connection":
        """Returns a new connection to the RabbitMQ server."""
        return await _establish_connection(self.url)

    async def _publish(self, message: QueueMessage) -> Any:
        """Publish message to the queue."""
        from aio_pika import DeliveryMode, ExchangeType, Message as AioPikaMessage

        message_type_str = message.type
        connection = await _establish_connection(self.url)

        async with connection:
            channel = await connection.channel()
            exchange = await channel.declare_exchange(
                self.exchange_name,
                ExchangeType.DIRECT,
            )
            message_body = json.dumps(message.model_dump()).encode("utf-8")

            aio_pika_message = AioPikaMessage(
                message_body,
                delivery_mode=DeliveryMode.PERSISTENT,
            )
            # Sending the message
            await exchange.publish(aio_pika_message, routing_key=message_type_str)
            logger.info(f"published message {message.id_}")

    async def register_consumer(
        self, consumer: BaseMessageQueueConsumer
    ) -> StartConsumingCallable:
        """Register a new consumer."""
        from aio_pika import ExchangeType

        connection = await _establish_connection(self.url)
        async with connection:
            channel = cast(Channel, await connection.channel())
            exchange = await channel.declare_exchange(
                self.exchange_name,
                ExchangeType.DIRECT,
            )
            queue = cast(Queue, await channel.declare_queue(name=consumer.message_type))
            await queue.bind(exchange)

        logger.info(
            f"Registered consumer {consumer.id_}: {consumer.message_type}",
        )

        async def start_consuming_callable() -> None:
            """StartConsumingCallable.

            Consumer of this queue, should call this in order to start consuming.
            """

            async def on_message(message: AbstractIncomingMessage) -> None:
                message = cast(IncomingMessage, message)
                async with message.process():
                    decoded_message = json.loads(message.body.decode("utf-8"))
                    queue_message = QueueMessage.model_validate(decoded_message)
                    await consumer.process_message(queue_message)

            connection = await _establish_connection(self.url)
            async with connection:
                channel = await connection.channel()
                exchange = await channel.declare_exchange(
                    self.exchange_name,
                    ExchangeType.DIRECT,
                )
                queue = cast(
                    Queue, await channel.declare_queue(name=consumer.message_type)
                )
                await queue.bind(exchange)

                await queue.consume(on_message)

                await asyncio.Future()

        return start_consuming_callable

    async def deregister_consumer(self, consumer: BaseMessageQueueConsumer) -> Any:
        """Deregister a consumer.

        Not implemented for this integration, as once the connection/channel is
        closed, the consumer is deregistered.
        """
        pass

    async def processing_loop(self) -> None:
        """A loop for getting messages from queues and sending to consumer.

        Not relevant for this class.
        """
        pass

    async def launch_local(self) -> asyncio.Task:
        """Launch the message queue locally, in-process.

        Launches a dummy task.
        """
        return asyncio.create_task(self.processing_loop())

    async def launch_server(self) -> None:
        """Launch the message queue server.

        Not relevant for this class. RabbitMQ server should already be launched.
        """
        pass

    async def cleanup_local(
        self, message_types: List[str], *args: Any, **kwargs: Dict[str, Any]
    ) -> None:
        """Perform any clean up of queues and exchanges."""
        connection = await self.new_connection()
        async with connection:
            channel = await connection.channel()
            for message_type in message_types:
                await channel.queue_delete(queue_name=message_type)
            await channel.exchange_delete(exchange_name=self.exchange_name)

    def as_config(self) -> BaseModel:
        return RabbitMQMessageQueueConfig(
            url=self.url, exchange_name=self.exchange_name
        )

from_url_params classmethod #

from_url_params(username: str, password: str, host: str, vhost: str = '', port: Optional[int] = None, secure: bool = False, exchange_name: str = DEFAULT_EXCHANGE_NAME) -> RabbitMQMessageQueue

Convenience constructor from url params.

Parameters:

Name Type Description Default
username str

username for the amqp authority

required
password str

password for the amqp authority

required
host str

host for rabbitmq server

required
port Optional[int]

port for rabbitmq server. Defaults to None.

None
secure bool

Whether or not to use SSL. Defaults to False.

False
exchange_name str

The exchange name. Defaults to DEFAULT_EXCHANGE_NAME.

DEFAULT_EXCHANGE_NAME

Returns:

Name Type Description
RabbitMQMessageQueue RabbitMQMessageQueue

A RabbitMQ MessageQueue integration.

Source code in llama_deploy/message_queues/rabbitmq.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@classmethod
def from_url_params(
    cls,
    username: str,
    password: str,
    host: str,
    vhost: str = "",
    port: Optional[int] = None,
    secure: bool = False,
    exchange_name: str = DEFAULT_EXCHANGE_NAME,
) -> "RabbitMQMessageQueue":
    """Convenience constructor from url params.

    Args:
        username (str): username for the amqp authority
        password (str): password for the amqp authority
        host (str): host for rabbitmq server
        port (Optional[int], optional): port for rabbitmq server. Defaults to None.
        secure (bool, optional): Whether or not to use SSL. Defaults to False.
        exchange_name (str, optional): The exchange name. Defaults to DEFAULT_EXCHANGE_NAME.

    Returns:
        RabbitMQMessageQueue: A RabbitMQ MessageQueue integration.
    """
    if not secure:
        if port:
            url = f"amqp://{username}:{password}@{host}:{port}/{vhost}"
        else:
            url = f"amqp://{username}:{password}@{host}/{vhost}"
    if secure:
        if port:
            url = f"amqps://{username}:{password}@{host}:{port}/{vhost}"
        else:
            url = f"amqps://{username}:{password}@{host}/{vhost}"
    return cls(url=url, exchange_name=exchange_name)

new_connection async #

new_connection() -> Connection

Returns a new connection to the RabbitMQ server.

Source code in llama_deploy/message_queues/rabbitmq.py
151
152
153
async def new_connection(self) -> "Connection":
    """Returns a new connection to the RabbitMQ server."""
    return await _establish_connection(self.url)

register_consumer async #

register_consumer(consumer: BaseMessageQueueConsumer) -> StartConsumingCallable

Register a new consumer.

Source code in llama_deploy/message_queues/rabbitmq.py
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def register_consumer(
    self, consumer: BaseMessageQueueConsumer
) -> StartConsumingCallable:
    """Register a new consumer."""
    from aio_pika import ExchangeType

    connection = await _establish_connection(self.url)
    async with connection:
        channel = cast(Channel, await connection.channel())
        exchange = await channel.declare_exchange(
            self.exchange_name,
            ExchangeType.DIRECT,
        )
        queue = cast(Queue, await channel.declare_queue(name=consumer.message_type))
        await queue.bind(exchange)

    logger.info(
        f"Registered consumer {consumer.id_}: {consumer.message_type}",
    )

    async def start_consuming_callable() -> None:
        """StartConsumingCallable.

        Consumer of this queue, should call this in order to start consuming.
        """

        async def on_message(message: AbstractIncomingMessage) -> None:
            message = cast(IncomingMessage, message)
            async with message.process():
                decoded_message = json.loads(message.body.decode("utf-8"))
                queue_message = QueueMessage.model_validate(decoded_message)
                await consumer.process_message(queue_message)

        connection = await _establish_connection(self.url)
        async with connection:
            channel = await connection.channel()
            exchange = await channel.declare_exchange(
                self.exchange_name,
                ExchangeType.DIRECT,
            )
            queue = cast(
                Queue, await channel.declare_queue(name=consumer.message_type)
            )
            await queue.bind(exchange)

            await queue.consume(on_message)

            await asyncio.Future()

    return start_consuming_callable

deregister_consumer async #

deregister_consumer(consumer: BaseMessageQueueConsumer) -> Any

Deregister a consumer.

Not implemented for this integration, as once the connection/channel is closed, the consumer is deregistered.

Source code in llama_deploy/message_queues/rabbitmq.py
229
230
231
232
233
234
235
async def deregister_consumer(self, consumer: BaseMessageQueueConsumer) -> Any:
    """Deregister a consumer.

    Not implemented for this integration, as once the connection/channel is
    closed, the consumer is deregistered.
    """
    pass

processing_loop async #

processing_loop() -> None

A loop for getting messages from queues and sending to consumer.

Not relevant for this class.

Source code in llama_deploy/message_queues/rabbitmq.py
237
238
239
240
241
242
async def processing_loop(self) -> None:
    """A loop for getting messages from queues and sending to consumer.

    Not relevant for this class.
    """
    pass

launch_local async #

launch_local() -> Task

Launch the message queue locally, in-process.

Launches a dummy task.

Source code in llama_deploy/message_queues/rabbitmq.py
244
245
246
247
248
249
async def launch_local(self) -> asyncio.Task:
    """Launch the message queue locally, in-process.

    Launches a dummy task.
    """
    return asyncio.create_task(self.processing_loop())

launch_server async #

launch_server() -> None

Launch the message queue server.

Not relevant for this class. RabbitMQ server should already be launched.

Source code in llama_deploy/message_queues/rabbitmq.py
251
252
253
254
255
256
async def launch_server(self) -> None:
    """Launch the message queue server.

    Not relevant for this class. RabbitMQ server should already be launched.
    """
    pass

cleanup_local async #

cleanup_local(message_types: List[str], *args: Any, **kwargs: Dict[str, Any]) -> None

Perform any clean up of queues and exchanges.

Source code in llama_deploy/message_queues/rabbitmq.py
258
259
260
261
262
263
264
265
266
267
async def cleanup_local(
    self, message_types: List[str], *args: Any, **kwargs: Dict[str, Any]
) -> None:
    """Perform any clean up of queues and exchanges."""
    connection = await self.new_connection()
    async with connection:
        channel = await connection.channel()
        for message_type in message_types:
            await channel.queue_delete(queue_name=message_type)
        await channel.exchange_delete(exchange_name=self.exchange_name)