Skip to content

RabbitDeclarerImpl

faststream.rabbit.helpers.declarer.RabbitDeclarerImpl #

RabbitDeclarerImpl(channel_manager: ChannelManager)

Bases: RabbitDeclarer

Source code in faststream/rabbit/helpers/declarer.py 
69
70
71
72
def __init__(self, channel_manager: "ChannelManager") -> None:
    self.__channel_manager = channel_manager
    self._queues: dict[RabbitQueue, aio_pika.RobustQueue] = {}
    self._exchanges: dict[RabbitExchange, aio_pika.RobustExchange] = {}

disconnect #

disconnect() -> None
Source code in faststream/rabbit/helpers/declarer.py 
77
78
79
def disconnect(self) -> None:
    self._queues.clear()
    self._exchanges.clear()

declare_queue async #

declare_queue(
    queue: RabbitQueue,
    declare: bool = EMPTY,
    *,
    channel: Optional[Channel] = None,
) -> RobustQueue
Source code in faststream/rabbit/helpers/declarer.py 
 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
async def declare_queue(
    self,
    queue: "RabbitQueue",
    declare: bool = EMPTY,
    *,
    channel: Optional["Channel"] = None,
) -> "aio_pika.RobustQueue":
    if (q := self._queues.get(queue)) is None:
        if declare is EMPTY:
            declare = queue.declare

        channel_obj = await self.__channel_manager.get_channel(channel)

        self._queues[queue] = q = cast(
            "aio_pika.RobustQueue",
            await channel_obj.declare_queue(
                name=queue.name,
                durable=queue.durable,
                exclusive=queue.exclusive,
                passive=not declare,
                auto_delete=queue.auto_delete,
                arguments=queue.arguments,
                timeout=queue.timeout,
                robust=queue.robust,
            ),
        )

    return q

declare_exchange async #

declare_exchange(
    exchange: RabbitExchange,
    declare: bool = EMPTY,
    *,
    channel: Optional[Channel] = None,
) -> RobustExchange
Source code in faststream/rabbit/helpers/declarer.py 
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
async def declare_exchange(
    self,
    exchange: "RabbitExchange",
    declare: bool = EMPTY,
    *,
    channel: Optional["Channel"] = None,
) -> "aio_pika.RobustExchange":
    channel_obj = await self.__channel_manager.get_channel(channel)

    if not exchange.name:
        return channel_obj.default_exchange

    if (exch := self._exchanges.get(exchange)) is None:
        if declare is EMPTY:
            declare = exchange.declare

        self._exchanges[exchange] = exch = cast(
            "aio_pika.RobustExchange",
            await channel_obj.declare_exchange(
                name=exchange.name,
                type=exchange.type.value,
                durable=exchange.durable,
                auto_delete=exchange.auto_delete,
                passive=not declare,
                arguments=exchange.arguments,
                timeout=exchange.timeout,
                robust=exchange.robust,
                internal=False,  # deprecated RMQ option
            ),
        )

        if exchange.bind_to is not None:
            parent = await self.declare_exchange(exchange.bind_to)
            await exch.bind(
                exchange=parent,
                routing_key=exchange.routing(),
                arguments=exchange.bind_arguments,
                timeout=exchange.timeout,
                robust=exchange.robust,
            )

    return exch