Declaring additional RabbitMQ queues

[bodograumann]

Hi,

I am trying to convert one of our services to use FastAPI and Propan.
We have one queue where we receive messages and another where we send results to.
From what I understand, only the exchanges and the queue that we consume from will be created by propan.
Can we add the possibility to

• create the queue that we are publishing to and
• create error queues for both queues?

Another thing is, that it seems like channel.declare_exchange is called everytime publish is used.
Shouldn't the exchange object be reusable?

[Lancetnik]

Well, you can create any queue at after_startup hook if you need. At native Propan you should get the channel from Context (or with propan.annotations.Channel). It's a native aio-pika channel object with you able to declare queue and exchange both this way

queue = await channel.declare_queue("hello", ...)

If you using Propan with FastAPI you can get the same channel from router

channel = router.broker._channel

If you want to declare some exchange/queue without hanler you should follow this way. I will work on more obvious way, but now it works too.

[bodograumann]

True, I was going to use broker._init_exchange as a workaround.
Using private attributes and methods is not a good solution in the long-run however.

Thanks for the tip about after_startup. That sounds like a good approach to make sure that the channel exists already.

[Lancetnik]

Another thing is, that it seems like channel.declare_exchange is called everytime publish is used.
Shouldn't the exchange object be reusable?

Now I am still thinking about the best way make it reusable. But for now aio_pika.declare_exchange check exchange existance at first, so it shouldn't be a big overkill.

[bodograumann]

Ah yeah, that should suffice for now :-)

[Lancetnik]

Anyway, appreciate your notes. So, I added Rabbit Broker.declare_queue, declare_exchange public methods and channel property. Also I added after_startup hook to all Propan FastAPI Routers . You can get all these changes with 0.1.2.17 update. It can be helpful in your case.