Skip to content

RabbitDeclarer

faststream.rabbit.helpers.declarer.RabbitDeclarer #

RabbitDeclarer(channel_manage)

An utility class to declare RabbitMQ queues and exchanges.

Source code in faststream/rabbit/helpers/declarer.py 
21
22
23
24
def __init__(self, channel_manage: "ChannelManager") -> None:
    self.__channel_manager = channel_manage
    self.__queues: Dict[RabbitQueue, aio_pika.RobustQueue] = {}
    self.__exchanges: Dict[RabbitExchange, aio_pika.RobustExchange] = {}

declare_queue async #

declare_queue(
    queue, declare=EMPTY, passive=EMPTY, *, channel=None
)

Declare a queue.

Source code in faststream/rabbit/helpers/declarer.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def declare_queue(
    self,
    queue: "RabbitQueue",
    declare: bool = EMPTY,
    passive: Annotated[
        bool,
        deprecated("Use `declare` instead. Will be removed in the 0.6.0 release."),
    ] = EMPTY,
    *,
    channel: Optional["Channel"] = None,
) -> "aio_pika.RobustQueue":
    """Declare a queue."""
    if (q := self.__queues.get(queue)) is None:
        channel_obj = await self.__channel_manager.get_channel(channel)
        if passive is not EMPTY:
            warnings.warn(
                DeprecationWarning(
                    "Use `declare` instead. Will be removed in the 0.6.0 release.",
                ),
                stacklevel=2,
            )
            declare = not passive
        elif declare is EMPTY:
            declare = queue.declare

        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, declare=EMPTY, passive=EMPTY, *, channel=None
)

Declare an exchange, parent exchanges and bind them each other.

Source code in faststream/rabbit/helpers/declarer.py 
 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
async def declare_exchange(
    self,
    exchange: "RabbitExchange",
    declare: bool = EMPTY,
    passive: Annotated[
        bool,
        deprecated("Use `declare` instead. Will be removed in the 0.6.0 release."),
    ] = EMPTY,
    *,
    channel: Optional["Channel"] = None,
) -> "aio_pika.RobustExchange":
    """Declare an exchange, parent exchanges and bind them each other."""
    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 passive is not EMPTY:
            warnings.warn(
                DeprecationWarning(
                    "Use `declare` instead. Will be removed in the 0.6.0 release.",
                ),
                stacklevel=2,
            )
            declare = not passive
        elif 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