15763470272

Committed 19 Jun 2025 05:36PM UTC coverage: 91.697% (-0.02%) from 91.72%

Build # 15763470272

Build Type

Pull #678

github

Committed by

web-flow

Commit Message

Merge 413dfe58e into c281314c6

Pull Request Pull Request #678: Another way of fixing the exchange issue

Coverage Stats

691 of 804 branches covered (85.95%)

6 of 7 new or added lines in 1 file covered. (85.71%)

1999 of 2180 relevant lines covered (91.7%)

3.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.23

/aio_pika/exchange.py

from typing import Optional, Union

import aiormq
from pamqp.common import Arguments

from .abc import (
    AbstractChannel, AbstractExchange, AbstractMessage, ExchangeParamType,
    ExchangeType, TimeoutType, get_exchange_name,
)
from .log import get_logger


log = get_logger(__name__)


class Exchange(AbstractExchange):
    """ Exchange abstraction """
    channel: AbstractChannel

    def __init__(
        self,
        channel: AbstractChannel,
        name: str,
        type: Union[ExchangeType, str] = ExchangeType.DIRECT,
        *,
        auto_delete: bool = False,
        durable: bool = False,
        internal: bool = False,
        passive: bool = False,
        arguments: Arguments = None,
    ):
        self._type = type.value if isinstance(type, ExchangeType) else type
        self.channel = channel
        self.name = name
        self.auto_delete = auto_delete
        self.durable = durable
        self.internal = internal
        self.passive = passive
        self.arguments = frozenset(arguments.items()) if arguments else frozenset()

    def __str__(self) -> str:
        return self.name

    def __repr__(self) -> str:
        return (
            f"<{self.__class__.__name__}({self}):"
            f" auto_delete={self.auto_delete},"
            f" durable={self.durable},"
            f" arguments={dict(self.arguments)!r})>"
        )

    def __eq__(self, other: object) -> bool:
        """
        Defines equality for Exchange objects.
        Two exchanges are considered equal if their name, type, and
        all boolean flags, and arguments are the same.
        """
        if not isinstance(other, Exchange):
            return False
        return (
            self.name == other.name and
            self._type == other._type and
            self.auto_delete == other.auto_delete and
            self.durable == other.durable and
            self.internal == other.internal and
            self.passive == other.passive and
            self.arguments == other.arguments
        )

    def __hash__(self) -> int:
        """
        Computes a hash for the Exchange object.
        The hash is based on the same attributes used in __eq__.
        """
        return hash((
            self.name,
            self._type,
            self.auto_delete,
            self.durable,
            self.internal,
            self.passive,
            self.arguments  # This is a frozenset, which is hashable
        ))

    async def declare(
        self, timeout: TimeoutType = None,
    ) -> aiormq.spec.Exchange.DeclareOk:
        channel = await self.channel.get_underlay_channel()
        return await channel.exchange_declare(
            self.name,
            exchange_type=self._type,
            durable=self.durable,
            auto_delete=self.auto_delete,
            internal=self.internal,
            passive=self.passive,
            arguments=dict(self.arguments),
            timeout=timeout,
        )

    async def bind(
        self,
        exchange: ExchangeParamType,
        routing_key: str = "",
        *,
        arguments: Arguments = None,
        timeout: TimeoutType = None,
    ) -> aiormq.spec.Exchange.BindOk:

        """ A binding can also be a relationship between two exchanges.
        This can be simply read as: this exchange is interested in messages
        from another exchange.

        Bindings can take an extra routing_key parameter. To avoid the confusion
        with a basic_publish parameter we're going to call it a binding key.

        .. code-block:: python

            client = await connect()

            routing_key = 'simple_routing_key'
            src_exchange_name = "source_exchange"
            dest_exchange_name = "destination_exchange"

            channel = await client.channel()
            src_exchange = await channel.declare_exchange(
                src_exchange_name, auto_delete=True
            )
            dest_exchange = await channel.declare_exchange(
                dest_exchange_name, auto_delete=True
            )
            queue = await channel.declare_queue(auto_delete=True)

            await queue.bind(dest_exchange, routing_key)
            await dest_exchange.bind(src_exchange, routing_key)

        :param exchange: :class:`aio_pika.exchange.Exchange` instance
        :param routing_key: routing key
        :param arguments: additional arguments
        :param timeout: execution timeout
        :return: :class:`None`
        """

        log.debug(
            "Binding exchange %r to exchange %r, routing_key=%r, arguments=%r",
            self,
            exchange,
            routing_key,
            arguments,
        )

        channel = await self.channel.get_underlay_channel()
        return await channel.exchange_bind(
            arguments=arguments,
            destination=self.name,
            routing_key=routing_key,
            source=get_exchange_name(exchange),
            timeout=timeout,
        )

    async def unbind(
        self,
        exchange: ExchangeParamType,
        routing_key: str = "",
        arguments: Arguments = None,
        timeout: TimeoutType = None,
    ) -> aiormq.spec.Exchange.UnbindOk:

        """ Remove exchange-to-exchange binding for this
        :class:`Exchange` instance

        :param exchange: :class:`aio_pika.exchange.Exchange` instance
        :param routing_key: routing key
        :param arguments: additional arguments
        :param timeout: execution timeout
        :return: :class:`None`
        """

        log.debug(
            "Unbinding exchange %r from exchange %r, "
            "routing_key=%r, arguments=%r",
            self,
            exchange,
            routing_key,
            arguments,
        )

        channel = await self.channel.get_underlay_channel()
        return await channel.exchange_unbind(
            arguments=arguments,
            destination=self.name,
            routing_key=routing_key,
            source=get_exchange_name(exchange),
            timeout=timeout,
        )

    async def publish(
        self,
        message: AbstractMessage,
        routing_key: str,
        *,
        mandatory: bool = True,
        immediate: bool = False,
        timeout: TimeoutType = None,
    ) -> Optional[aiormq.abc.ConfirmationFrameType]:

        """ Publish the message to the queue. `aio-pika` uses
        `publisher confirms`_ extension for message delivery.

        .. _publisher confirms: https://www.rabbitmq.com/confirms.html

        """

        log.debug(
            "Publishing message with routing key %r via exchange %r: %r",
            routing_key,
            self,
            message,
        )

        if self.internal:
            # Caught on the client side to prevent channel closure
            raise ValueError(
                f"Can not publish to internal exchange: '{self.name}'!",
            )

        if self.channel.is_closed:
            raise aiormq.exceptions.ChannelInvalidStateError(
                "%r closed" % self.channel,
            )

        channel = await self.channel.get_underlay_channel()
        return await channel.basic_publish(
            exchange=self.name,
            routing_key=routing_key,
            body=message.body,
            properties=message.properties,
            mandatory=mandatory,
            immediate=immediate,
            timeout=timeout,
        )

    async def delete(
        self, if_unused: bool = False, timeout: TimeoutType = None,
    ) -> aiormq.spec.Exchange.DeleteOk:

        """ Delete the queue

        :param timeout: operation timeout
        :param if_unused: perform deletion when queue has no bindings.
        """

        log.info("Deleting %r", self)
        channel = await self.channel.get_underlay_channel()
        result = await channel.exchange_delete(
            self.name, if_unused=if_unused, timeout=timeout,
        )
        del self.channel
        return result


__all__ = ("Exchange", "ExchangeType", "ExchangeParamType")

1	from typing import Optional, Union	4✔
2
3	import aiormq	4✔
4	from pamqp.common import Arguments	4✔
5
6	from .abc import (	4✔
7	AbstractChannel, AbstractExchange, AbstractMessage, ExchangeParamType,
8	ExchangeType, TimeoutType, get_exchange_name,
9	)
10	from .log import get_logger	4✔
11
12
13	log = get_logger(__name__)	4✔
14
15
16	class Exchange(AbstractExchange):	4✔
17	""" Exchange abstraction """
18	channel: AbstractChannel	4✔
19
20	def __init__(	4✔
21	self,
22	channel: AbstractChannel,
23	name: str,
24	type: Union[ExchangeType, str] = ExchangeType.DIRECT,
25	*,
26	auto_delete: bool = False,
27	durable: bool = False,
28	internal: bool = False,
29	passive: bool = False,
30	arguments: Arguments = None,
31	):
32	self._type = type.value if isinstance(type, ExchangeType) else type	4✔
33	self.channel = channel	4✔
34	self.name = name	4✔
35	self.auto_delete = auto_delete	4✔
36	self.durable = durable	4✔
37	self.internal = internal	4✔
38	self.passive = passive	4✔
39	self.arguments = frozenset(arguments.items()) if arguments else frozenset()	4✔
40
41	def __str__(self) -> str:	4✔
42	return self.name	4✔
43
44	def __repr__(self) -> str:	4✔
45	return (	4✔
46	f"<{self.__class__.__name__}({self}):"
47	f" auto_delete={self.auto_delete},"
48	f" durable={self.durable},"
49	f" arguments={dict(self.arguments)!r})>"
50	)
51
52	def __eq__(self, other: object) -> bool:	4✔
53	"""
54	Defines equality for Exchange objects.
55	Two exchanges are considered equal if their name, type, and
56	all boolean flags, and arguments are the same.
57	"""
58	if not isinstance(other, Exchange):	4!
NEW 59	return False	×
60	return (	4✔
61	self.name == other.name and
62	self._type == other._type and
63	self.auto_delete == other.auto_delete and
64	self.durable == other.durable and
65	self.internal == other.internal and
66	self.passive == other.passive and
67	self.arguments == other.arguments
68	)
69
70	def __hash__(self) -> int:	4✔
71	"""
72	Computes a hash for the Exchange object.
73	The hash is based on the same attributes used in __eq__.
74	"""
75	return hash((	4✔
76	self.name,
77	self._type,
78	self.auto_delete,
79	self.durable,
80	self.internal,
81	self.passive,
82	self.arguments # This is a frozenset, which is hashable
83	))
84
85	async def declare(	4✔
86	self, timeout: TimeoutType = None,
87	) -> aiormq.spec.Exchange.DeclareOk:
88	channel = await self.channel.get_underlay_channel()	4✔
89	return await channel.exchange_declare(	4✔
90	self.name,
91	exchange_type=self._type,
92	durable=self.durable,
93	auto_delete=self.auto_delete,
94	internal=self.internal,
95	passive=self.passive,
96	arguments=dict(self.arguments),
97	timeout=timeout,
98	)
99
100	async def bind(	4✔
101	self,
102	exchange: ExchangeParamType,
103	routing_key: str = "",
104	*,
105	arguments: Arguments = None,
106	timeout: TimeoutType = None,
107	) -> aiormq.spec.Exchange.BindOk:
108
109	""" A binding can also be a relationship between two exchanges.
110	This can be simply read as: this exchange is interested in messages
111	from another exchange.
112
113	Bindings can take an extra routing_key parameter. To avoid the confusion
114	with a basic_publish parameter we're going to call it a binding key.
115
116	.. code-block:: python
117
118	client = await connect()
119
120	routing_key = 'simple_routing_key'
121	src_exchange_name = "source_exchange"
122	dest_exchange_name = "destination_exchange"
123
124	channel = await client.channel()
125	src_exchange = await channel.declare_exchange(
126	src_exchange_name, auto_delete=True
127	)
128	dest_exchange = await channel.declare_exchange(
129	dest_exchange_name, auto_delete=True
130	)
131	queue = await channel.declare_queue(auto_delete=True)
132
133	await queue.bind(dest_exchange, routing_key)
134	await dest_exchange.bind(src_exchange, routing_key)
135
136	:param exchange: :class:`aio_pika.exchange.Exchange` instance
137	:param routing_key: routing key
138	:param arguments: additional arguments
139	:param timeout: execution timeout
140	:return: :class:`None`
141	"""
142
143	log.debug(	4✔
144	"Binding exchange %r to exchange %r, routing_key=%r, arguments=%r",
145	self,
146	exchange,
147	routing_key,
148	arguments,
149	)
150
151	channel = await self.channel.get_underlay_channel()	4✔
152	return await channel.exchange_bind(	4✔
153	arguments=arguments,
154	destination=self.name,
155	routing_key=routing_key,
156	source=get_exchange_name(exchange),
157	timeout=timeout,
158	)
159
160	async def unbind(	4✔
161	self,
162	exchange: ExchangeParamType,
163	routing_key: str = "",
164	arguments: Arguments = None,
165	timeout: TimeoutType = None,
166	) -> aiormq.spec.Exchange.UnbindOk:
167
168	""" Remove exchange-to-exchange binding for this
169	:class:`Exchange` instance
170
171	:param exchange: :class:`aio_pika.exchange.Exchange` instance
172	:param routing_key: routing key
173	:param arguments: additional arguments
174	:param timeout: execution timeout
175	:return: :class:`None`
176	"""
177
178	log.debug(	4✔
179	"Unbinding exchange %r from exchange %r, "
180	"routing_key=%r, arguments=%r",
181	self,
182	exchange,
183	routing_key,
184	arguments,
185	)
186
187	channel = await self.channel.get_underlay_channel()	4✔
188	return await channel.exchange_unbind(	4✔
189	arguments=arguments,
190	destination=self.name,
191	routing_key=routing_key,
192	source=get_exchange_name(exchange),
193	timeout=timeout,
194	)
195
196	async def publish(	4✔
197	self,
198	message: AbstractMessage,
199	routing_key: str,
200	*,
201	mandatory: bool = True,
202	immediate: bool = False,
203	timeout: TimeoutType = None,
204	) -> Optional[aiormq.abc.ConfirmationFrameType]:
205
206	""" Publish the message to the queue. `aio-pika` uses
207	`publisher confirms`_ extension for message delivery.
208
209	.. _publisher confirms: https://www.rabbitmq.com/confirms.html
210
211	"""
212
213	log.debug(	4✔
214	"Publishing message with routing key %r via exchange %r: %r",
215	routing_key,
216	self,
217	message,
218	)
219
220	if self.internal:	4!
221	# Caught on the client side to prevent channel closure
222	raise ValueError(	×
223	f"Can not publish to internal exchange: '{self.name}'!",
224	)
225
226	if self.channel.is_closed:	4✔
227	raise aiormq.exceptions.ChannelInvalidStateError(	4✔
228	"%r closed" % self.channel,
229	)
230
231	channel = await self.channel.get_underlay_channel()	4✔
232	return await channel.basic_publish(	4✔
233	exchange=self.name,
234	routing_key=routing_key,
235	body=message.body,
236	properties=message.properties,
237	mandatory=mandatory,
238	immediate=immediate,
239	timeout=timeout,
240	)
241
242	async def delete(	4✔
243	self, if_unused: bool = False, timeout: TimeoutType = None,
244	) -> aiormq.spec.Exchange.DeleteOk:
245
246	""" Delete the queue
247
248	:param timeout: operation timeout
249	:param if_unused: perform deletion when queue has no bindings.
250	"""
251
252	log.info("Deleting %r", self)	4✔
253	channel = await self.channel.get_underlay_channel()	4✔
254	result = await channel.exchange_delete(	4✔
255	self.name, if_unused=if_unused, timeout=timeout,
256	)
257	del self.channel	4✔
258	return result	4✔
259
260
261	__all__ = ("Exchange", "ExchangeType", "ExchangeParamType")	4✔

mosquito / aio-pika / 15763470272

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous