#3241

pending completion

Build # #3241

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

<a href="https://github.com/mosquito/aio-pika/commit/<a class=hub.com/mosquito/aio-pika/commit/32b60ce7762cf9fd79c65b3da08f2ad6f4a6a483">32b60ce77<a href="https://github.com/mosquito/aio-pika/commit/32b60ce7762cf9fd79c65b3da08f2ad6f4a6a483">">Merge </a><a class="double-link" href="https://github.com/mosquito/aio-pika/commit/<a class="double-link" href="https://github.com/mosquito/aio-pika/commit/a069a03518ff3ef86064e2f37ceb3d3b6de09230">a069a0351</a>">a069a0351</a><a href="https://github.com/mosquito/aio-pika/commit/32b60ce7762cf9fd79c65b3da08f2ad6f4a6a483"> into b6631d426">b6631d426</a>

Run Details

347 of 433 branches covered (80.14%)

4 of 4 new or added lines in 1 file covered. (100.0%)

1768 of 1981 relevant lines covered (89.25%)

0.89 hits per line

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

90.24

/aio_pika/connection.py

import asyncio
import logging
from functools import partial
from types import TracebackType
from typing import Any, Callable, Dict, Optional, Tuple, Type, TypeVar, Union
from warnings import warn

import aiormq
import aiormq.abc
from aiormq.tools import censor_url
from pamqp.common import FieldTable
from yarl import URL

# This needed only for migration from 6.x to 7.x
# TODO: Remove this in 8.x release
from .abc import ConnectionType  # noqa
from .abc import (
    AbstractChannel, AbstractConnection, ConnectionCloseCallback, TimeoutType,
)
from .channel import Channel
from .tools import CallbackCollection


log = logging.getLogger(__name__)
T = TypeVar("T")


class Connection(AbstractConnection):
    """ Connection abstraction """

    CHANNEL_CLASS: Type[Channel] = Channel
    KWARGS_TYPES: Tuple[Tuple[str, Callable[[str], Any], str], ...] = ()

    @property
    def is_closed(self) -> bool:
        return self.closing.done()

    async def close(
        self, exc: Optional[aiormq.abc.ExceptionType] = asyncio.CancelledError,
    ) -> None:
        if not self.closing.done():
            self.closing.set_result(exc)

        if not hasattr(self, "connection"):
            return None

        await self.connection.close(exc)

        if hasattr(self, "connection"):
            del self.connection

    @classmethod
    def _parse_kwargs(cls, kwargs: Dict[str, Any]) -> Dict[str, Any]:
        result = {}
        for key, parser, default in cls.KWARGS_TYPES:
            result[key] = parser(kwargs.get(key, default))
        return result

    def __init__(
        self, url: URL, loop: Optional[asyncio.AbstractEventLoop] = None,
        **kwargs: Any
    ):
        super().__init__(url, loop, **kwargs)

        self.loop = loop or asyncio.get_event_loop()
        self.url = URL(url)

        self.kwargs: Dict[str, Any] = self._parse_kwargs(
            kwargs or dict(self.url.query),
        )

        self.connection: aiormq.abc.AbstractConnection
        self.close_callbacks = CallbackCollection(self)
        self.connected: asyncio.Event = asyncio.Event()
        self.closing: asyncio.Future = self.loop.create_future()

    def __str__(self) -> str:
        return str(censor_url(self.url))

    def __repr__(self) -> str:
        return f'<{self.__class__.__name__}: "{self}">'

    def add_close_callback(
        self, callback: ConnectionCloseCallback, weak: bool = False,
    ) -> None:
        warn(
            "This method will be removed from future release. "
            f"Use {self.__class__.__name__}.close_callbacks.add instead",
            DeprecationWarning,
            stacklevel=2,
        )
        self.close_callbacks.add(callback, weak=weak)

    def _on_connection_close(
        self, connection: aiormq.abc.AbstractConnection,
        closing: asyncio.Future,
    ) -> None:
        log.debug("Closing AMQP connection %r", connection)
        exc: Optional[BaseException] = closing.exception()
        self.close_callbacks(exc)

        if self.closing.done():
            return

        if exc is not None:
            self.closing.set_exception(exc)
            return

        self.closing.set_result(closing.result())

    async def _make_connection(
        self, *, timeout: TimeoutType = None, **kwargs: Any
    ) -> aiormq.abc.AbstractConnection:
        connection: aiormq.abc.AbstractConnection = await asyncio.wait_for(
            aiormq.connect(self.url, **kwargs), timeout=timeout,
        )
        connection.closing.add_done_callback(
            partial(self._on_connection_close, connection),
        )
        await connection.ready()
        return connection

    async def connect(
        self, timeout: TimeoutType = None, **kwargs: Any
    ) -> None:
        """ Connect to AMQP server. This method should be called after
        :func:`aio_pika.connection.Connection.__init__`

        .. note::
            This method is called by :func:`connect`.
            You shouldn't call it explicitly.

        """
        self.connection = (
            await self._make_connection(timeout=timeout, **kwargs)
        )
        self.connected.set()
        self.connection.closing.add_done_callback(
            lambda _: self.connected.clear(),
        )

    def channel(
        self,
        channel_number: int = None,
        publisher_confirms: bool = True,
        on_return_raises: bool = False,
    ) -> AbstractChannel:
        """ Coroutine which returns new instance of :class:`Channel`.

        Example:

        .. code-block:: python

            import aio_pika

            async def main(loop):
                connection = await aio_pika.connect(
                    "amqp://guest:guest@127.0.0.1/"
                )

                channel1 = connection.channel()
                await channel1.close()

                # Creates channel with specific channel number
                channel42 = connection.channel(42)
                await channel42.close()

                # For working with transactions
                channel_no_confirms = connection.channel(
                    publisher_confirms=True
                )
                await channel_no_confirms.close()

        Also available as an asynchronous context manager:

        .. code-block:: python

            import aio_pika

            async def main(loop):
                connection = await aio_pika.connect(
                    "amqp://guest:guest@127.0.0.1/"
                )

                async with connection.channel() as channel:
                    # channel is open and available

                # channel is now closed

        :param channel_number: specify the channel number explicit
        :param publisher_confirms:
            if `True` the :func:`aio_pika.Exchange.publish` method will be
            return :class:`bool` after publish is complete. Otherwise the
            :func:`aio_pika.Exchange.publish` method will be return
            :class:`None`
        :param on_return_raises:
            raise an :class:`aio_pika.exceptions.DeliveryError`
            when mandatory message will be returned
        """

        log.debug("Creating AMQP channel for connection: %r", self)

        channel = self.CHANNEL_CLASS(
            connection=self,
            channel_number=channel_number,
            publisher_confirms=publisher_confirms,
            on_return_raises=on_return_raises,
        )

        log.debug("Channel created: %r", channel)
        return channel

    async def ready(self) -> None:
        await self.connected.wait()

    def __del__(self) -> None:
        if (
            self.is_closed or
            self.loop.is_closed() or
            not hasattr(self, "connection")
        ):
            return

        asyncio.shield(self.close())

    async def __aenter__(self) -> "Connection":
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        await self.close()


def make_url(
    url: Union[str, URL] = None,
    *,
    host: str = "localhost",
    port: int = 5672,
    login: str = "guest",
    password: str = "guest",
    virtualhost: str = "/",
    ssl: bool = False,
    ssl_options: dict = None,
    **kwargs: Any
) -> URL:
    if url is not None:
        if not isinstance(url, URL):
            return URL(url)
        return url

    kw = kwargs
    kw.update(ssl_options or {})

    return URL.build(
        scheme="amqps" if ssl else "amqp",
        host=host,
        port=port,
        user=login,
        password=password,
        # yarl >= 1.3.0 requires path beginning with slash
        path="/" + virtualhost,
        query=kw,
    )


async def connect(
    url: Union[str, URL] = None,
    *,
    host: str = "localhost",
    port: int = 5672,
    login: str = "guest",
    password: str = "guest",
    virtualhost: str = "/",
    ssl: bool = False,
    loop: asyncio.AbstractEventLoop = None,
    ssl_options: dict = None,
    timeout: TimeoutType = None,
    client_properties: FieldTable = None,
    connection_class: Type[AbstractConnection] = Connection,
    **kwargs: Any
) -> AbstractConnection:

    """ Make connection to the broker.

    Example:

    .. code-block:: python

        import aio_pika

        async def main():
            connection = await aio_pika.connect(
                "amqp://guest:guest@127.0.0.1/"
            )

    Connect to localhost with default credentials:

    .. code-block:: python

        import aio_pika

        async def main():
            connection = await aio_pika.connect()

    .. note::

        The available keys for ssl_options parameter are:
            * cert_reqs
            * certfile
            * keyfile
            * ssl_version

        For an information on what the ssl_options can be set to reference the
        `official Python documentation`_ .

    Set connection name for RabbitMQ admin panel:

    .. code-block:: python

        read_connection = await connect(
            client_properties={
                'connection_name': 'Read connection'
            }
        )

        write_connection = await connect(
            client_properties={
                'connection_name': 'Write connection'
            }
        )

    .. note:

        ``client_properties`` argument requires ``aiormq>=2.9``

    URL string might be contain ssl parameters e.g.
    `amqps://user:pass@host//?ca_certs=ca.pem&certfile=crt.pem&keyfile=key.pem`

    :param client_properties: add custom client capability.
    :param url:
        RFC3986_ formatted broker address. When :class:`None`
        will be used keyword arguments.
    :param host: hostname of the broker
    :param port: broker port 5672 by default
    :param login: username string. `'guest'` by default.
    :param password: password string. `'guest'` by default.
    :param virtualhost: virtualhost parameter. `'/'` by default
    :param ssl: use SSL for connection. Should be used with addition kwargs.
    :param ssl_options: A dict of values for the SSL connection.
    :param timeout: connection timeout in seconds
    :param loop:
        Event loop (:func:`asyncio.get_event_loop()` when :class:`None`)
    :param connection_class: Factory of a new connection
    :param kwargs: addition parameters which will be passed to the connection.
    :return: :class:`aio_pika.connection.Connection`

    .. _RFC3986: https://goo.gl/MzgYAs
    .. _official Python documentation: https://goo.gl/pty9xA


    """

    connection: AbstractConnection = connection_class(
        make_url(
            url,
            host=host,
            port=port,
            login=login,
            password=password,
            virtualhost=virtualhost,
            ssl=ssl,
            ssl_options=ssl_options,
            **kwargs
        ),
        loop=loop,
    )

    await connection.connect(
        timeout=timeout,
        client_properties=client_properties,
        loop=loop,
    )

    return connection


__all__ = ("connect", "Connection")

1	import asyncio	1✔
2	import logging	1✔
3	from functools import partial	1✔
4	from types import TracebackType	1✔
5	from typing import Any, Callable, Dict, Optional, Tuple, Type, TypeVar, Union	1✔
6	from warnings import warn	1✔
7
8	import aiormq	1✔
9	import aiormq.abc	1✔
10	from aiormq.tools import censor_url	1✔
11	from pamqp.common import FieldTable	1✔
12	from yarl import URL	1✔
13
14	# This needed only for migration from 6.x to 7.x
15	# TODO: Remove this in 8.x release
16	from .abc import ConnectionType # noqa	1✔
17	from .abc import (	1✔
18	AbstractChannel, AbstractConnection, ConnectionCloseCallback, TimeoutType,
19	)
20	from .channel import Channel	1✔
21	from .tools import CallbackCollection	1✔
22
23
24	log = logging.getLogger(__name__)	1✔
25	T = TypeVar("T")	1✔
26
27
28	class Connection(AbstractConnection):	1✔
29	""" Connection abstraction """
30
31	CHANNEL_CLASS: Type[Channel] = Channel	1✔
32	KWARGS_TYPES: Tuple[Tuple[str, Callable[[str], Any], str], ...] = ()	1✔
33
34	@property	1✔
35	def is_closed(self) -> bool:	1✔
36	return self.closing.done()	1✔
37
38	async def close(	1✔
39	self, exc: Optional[aiormq.abc.ExceptionType] = asyncio.CancelledError,
40	) -> None:
41	if not self.closing.done():	1✔
42	self.closing.set_result(exc)	1✔
43
44	if not hasattr(self, "connection"):	1✔
45	return None	1✔
46
47	await self.connection.close(exc)	1✔
48
49	if hasattr(self, "connection"):	1✔
50	del self.connection	1✔
51
52	@classmethod	1✔
53	def _parse_kwargs(cls, kwargs: Dict[str, Any]) -> Dict[str, Any]:	1✔
54	result = {}	1✔
55	for key, parser, default in cls.KWARGS_TYPES:	1✔
56	result[key] = parser(kwargs.get(key, default))	1✔
57	return result	1✔
58
59	def __init__(	1✔
60	self, url: URL, loop: Optional[asyncio.AbstractEventLoop] = None,
61	**kwargs: Any
62	):
63	super().__init__(url, loop, **kwargs)	1✔
64
65	self.loop = loop or asyncio.get_event_loop()	1✔
66	self.url = URL(url)	1✔
67
68	self.kwargs: Dict[str, Any] = self._parse_kwargs(	1✔
69	kwargs or dict(self.url.query),
70	)
71
72	self.connection: aiormq.abc.AbstractConnection	1✔
73	self.close_callbacks = CallbackCollection(self)	1✔
74	self.connected: asyncio.Event = asyncio.Event()	1✔
75	self.closing: asyncio.Future = self.loop.create_future()	1✔
76
77	def __str__(self) -> str:	1✔
78	return str(censor_url(self.url))	1✔
79
80	def __repr__(self) -> str:	1✔
81	return f'<{self.__class__.__name__}: "{self}">'	×
82
83	def add_close_callback(	1✔
84	self, callback: ConnectionCloseCallback, weak: bool = False,
85	) -> None:
86	warn(	×
87	"This method will be removed from future release. "
88	f"Use {self.__class__.__name__}.close_callbacks.add instead",
89	DeprecationWarning,
90	stacklevel=2,
91	)
92	self.close_callbacks.add(callback, weak=weak)	×
93
94	def _on_connection_close(	1✔
95	self, connection: aiormq.abc.AbstractConnection,
96	closing: asyncio.Future,
97	) -> None:
98	log.debug("Closing AMQP connection %r", connection)	1✔
99	exc: Optional[BaseException] = closing.exception()	1✔
100	self.close_callbacks(exc)	1✔
101
102	if self.closing.done():	1!
103	return	1✔
104
105	if exc is not None:	×
106	self.closing.set_exception(exc)	×
107	return	×
108
109	self.closing.set_result(closing.result())	×
110
111	async def _make_connection(	1✔
112	self, , timeout: TimeoutType = None, *kwargs: Any
113	) -> aiormq.abc.AbstractConnection:
114	connection: aiormq.abc.AbstractConnection = await asyncio.wait_for(	1✔
115	aiormq.connect(self.url, **kwargs), timeout=timeout,
116	)
117	connection.closing.add_done_callback(	1✔
118	partial(self._on_connection_close, connection),
119	)
120	await connection.ready()	1✔
121	return connection	1✔
122
123	async def connect(	1✔
124	self, timeout: TimeoutType = None, **kwargs: Any
125	) -> None:
126	""" Connect to AMQP server. This method should be called after
127	:func:`aio_pika.connection.Connection.__init__`
128
129	.. note::
130	This method is called by :func:`connect`.
131	You shouldn't call it explicitly.
132
133	"""
134	self.connection = (	1✔
135	await self._make_connection(timeout=timeout, **kwargs)
136	)
137	self.connected.set()	1✔
138	self.connection.closing.add_done_callback(	1✔
139	lambda _: self.connected.clear(),
140	)
141
142	def channel(	1✔
143	self,
144	channel_number: int = None,
145	publisher_confirms: bool = True,
146	on_return_raises: bool = False,
147	) -> AbstractChannel:
148	""" Coroutine which returns new instance of :class:`Channel`.
149
150	Example:
151
152	.. code-block:: python
153
154	import aio_pika
155
156	async def main(loop):
157	connection = await aio_pika.connect(
158	"amqp://guest:guest@127.0.0.1/"
159	)
160
161	channel1 = connection.channel()
162	await channel1.close()
163
164	# Creates channel with specific channel number
165	channel42 = connection.channel(42)
166	await channel42.close()
167
168	# For working with transactions
169	channel_no_confirms = connection.channel(
170	publisher_confirms=True
171	)
172	await channel_no_confirms.close()
173
174	Also available as an asynchronous context manager:
175
176	.. code-block:: python
177
178	import aio_pika
179
180	async def main(loop):
181	connection = await aio_pika.connect(
182	"amqp://guest:guest@127.0.0.1/"
183	)
184
185	async with connection.channel() as channel:
186	# channel is open and available
187
188	# channel is now closed
189
190	:param channel_number: specify the channel number explicit
191	:param publisher_confirms:
192	if `True` the :func:`aio_pika.Exchange.publish` method will be
193	return :class:`bool` after publish is complete. Otherwise the
194	:func:`aio_pika.Exchange.publish` method will be return
195	:class:`None`
196	:param on_return_raises:
197	raise an :class:`aio_pika.exceptions.DeliveryError`
198	when mandatory message will be returned
199	"""
200
201	log.debug("Creating AMQP channel for connection: %r", self)	1✔
202
203	channel = self.CHANNEL_CLASS(	1✔
204	connection=self,
205	channel_number=channel_number,
206	publisher_confirms=publisher_confirms,
207	on_return_raises=on_return_raises,
208	)
209
210	log.debug("Channel created: %r", channel)	1✔
211	return channel	1✔
212
213	async def ready(self) -> None:	1✔
214	await self.connected.wait()	1✔
215
216	def __del__(self) -> None:	1✔
217	if (	1!
218	self.is_closed or
219	self.loop.is_closed() or
220	not hasattr(self, "connection")
221	):
222	return	1✔
223
224	asyncio.shield(self.close())	×
225
226	async def __aenter__(self) -> "Connection":	1✔
227	return self	1✔
228
229	async def __aexit__(	1✔
230	self,
231	exc_type: Optional[Type[BaseException]],
232	exc_val: Optional[BaseException],
233	exc_tb: Optional[TracebackType],
234	) -> None:
235	await self.close()	1✔
236
237
238	def make_url(	1✔
239	url: Union[str, URL] = None,
240	*,
241	host: str = "localhost",
242	port: int = 5672,
243	login: str = "guest",
244	password: str = "guest",
245	virtualhost: str = "/",
246	ssl: bool = False,
247	ssl_options: dict = None,
248	**kwargs: Any
249	) -> URL:
250	if url is not None:	1✔
251	if not isinstance(url, URL):	1✔
252	return URL(url)	1✔
253	return url	1✔
254
255	kw = kwargs	1✔
256	kw.update(ssl_options or {})	1✔
257
258	return URL.build(	1✔
259	scheme="amqps" if ssl else "amqp",
260	host=host,
261	port=port,
262	user=login,
263	password=password,
264	# yarl >= 1.3.0 requires path beginning with slash
265	path="/" + virtualhost,
266	query=kw,
267	)
268
269
270	async def connect(	1✔
271	url: Union[str, URL] = None,
272	*,
273	host: str = "localhost",
274	port: int = 5672,
275	login: str = "guest",
276	password: str = "guest",
277	virtualhost: str = "/",
278	ssl: bool = False,
279	loop: asyncio.AbstractEventLoop = None,
280	ssl_options: dict = None,
281	timeout: TimeoutType = None,
282	client_properties: FieldTable = None,
283	connection_class: Type[AbstractConnection] = Connection,
284	**kwargs: Any
285	) -> AbstractConnection:
286
287	""" Make connection to the broker.
288
289	Example:
290
291	.. code-block:: python
292
293	import aio_pika
294
295	async def main():
296	connection = await aio_pika.connect(
297	"amqp://guest:guest@127.0.0.1/"
298	)
299
300	Connect to localhost with default credentials:
301
302	.. code-block:: python
303
304	import aio_pika
305
306	async def main():
307	connection = await aio_pika.connect()
308
309	.. note::
310
311	The available keys for ssl_options parameter are:
312	* cert_reqs
313	* certfile
314	* keyfile
315	* ssl_version
316
317	For an information on what the ssl_options can be set to reference the
318	`official Python documentation`_ .
319
320	Set connection name for RabbitMQ admin panel:
321
322	.. code-block:: python
323
324	read_connection = await connect(
325	client_properties={
326	'connection_name': 'Read connection'
327	}
328	)
329
330	write_connection = await connect(
331	client_properties={
332	'connection_name': 'Write connection'
333	}
334	)
335
336	.. note:
337
338	``client_properties`` argument requires ``aiormq>=2.9``
339
340	URL string might be contain ssl parameters e.g.
341	`amqps://user:pass@host//?ca_certs=ca.pem&certfile=crt.pem&keyfile=key.pem`
342
343	:param client_properties: add custom client capability.
344	:param url:
345	RFC3986_ formatted broker address. When :class:`None`
346	will be used keyword arguments.
347	:param host: hostname of the broker
348	:param port: broker port 5672 by default
349	:param login: username string. `'guest'` by default.
350	:param password: password string. `'guest'` by default.
351	:param virtualhost: virtualhost parameter. `'/'` by default
352	:param ssl: use SSL for connection. Should be used with addition kwargs.
353	:param ssl_options: A dict of values for the SSL connection.
354	:param timeout: connection timeout in seconds
355	:param loop:
356	Event loop (:func:`asyncio.get_event_loop()` when :class:`None`)
357	:param connection_class: Factory of a new connection
358	:param kwargs: addition parameters which will be passed to the connection.
359	:return: :class:`aio_pika.connection.Connection`
360
361	.. _RFC3986: https://goo.gl/MzgYAs
362	.. _official Python documentation: https://goo.gl/pty9xA
363
364
365	"""
366
367	connection: AbstractConnection = connection_class(	1✔
368	make_url(
369	url,
370	host=host,
371	port=port,
372	login=login,
373	password=password,
374	virtualhost=virtualhost,
375	ssl=ssl,
376	ssl_options=ssl_options,
377	**kwargs
378	),
379	loop=loop,
380	)
381
382	await connection.connect(	1✔
383	timeout=timeout,
384	client_properties=client_properties,
385	loop=loop,
386	)
387
388	return connection	1✔
389
390
391	__all__ = ("connect", "Connection")	1✔

mosquito / aio-pika / #3241

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