Stop passing loop= parameter (deprecated in asyncio 3.8)

Author: 1st1

asyncpg/_testbase/__init__.py

@@ -80,7 +80,7 @@ def wrapper(self, *args, __meth__=meth, **kwargs):
                 coro = __meth__(self, *args, **kwargs)
                 timeout = getattr(__meth__, '__timeout__', mcls.TEST_TIMEOUT)
                 if timeout:
-                    coro = asyncio.wait_for(coro, timeout, loop=self.loop)
+                    coro = asyncio.wait_for(coro, timeout)
                     try:
                         self.loop.run_until_complete(coro)
                     except asyncio.TimeoutError:

asyncpg/_testbase/fuzzer.py

@@ -36,15 +36,13 @@ def __init__(self, *, listening_addr: str='127.0.0.1',
         self.listen_task = None
 
     async def _wait(self, work):
-        work_task = asyncio.ensure_future(work, loop=self.loop)
-        stop_event_task = asyncio.ensure_future(self.stop_event.wait(),
-                                                loop=self.loop)
+        work_task = asyncio.ensure_future(work)
+        stop_event_task = asyncio.ensure_future(self.stop_event.wait())
 
         try:
             await asyncio.wait(
                 [work_task, stop_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.stop_event.is_set():
                 raise StopServer()
@@ -58,7 +56,8 @@ async def _wait(self, work):
 
     def start(self):
         started = threading.Event()
-        self.thread = threading.Thread(target=self._start, args=(started,))
+        self.thread = threading.Thread(
+            target=self._start_thread, args=(started,))
         self.thread.start()
         if not started.wait(timeout=2):
             raise RuntimeError('fuzzer proxy failed to start')
@@ -70,13 +69,14 @@ def stop(self):
     def _stop(self):
         self.stop_event.set()
 
-    def _start(self, started_event):
+    def _start_thread(self, started_event):
         self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
 
-        self.connectivity = asyncio.Event(loop=self.loop)
+        self.connectivity = asyncio.Event()
         self.connectivity.set()
-        self.connectivity_loss = asyncio.Event(loop=self.loop)
-        self.stop_event = asyncio.Event(loop=self.loop)
+        self.connectivity_loss = asyncio.Event()
+        self.stop_event = asyncio.Event()
 
         if self.listening_port is None:
             self.listening_port = cluster.find_available_port()
@@ -92,15 +92,15 @@ def _start(self, started_event):
             self.loop.close()
 
     async def _main(self, started_event):
-        self.listen_task = asyncio.ensure_future(self.listen(), loop=self.loop)
+        self.listen_task = asyncio.ensure_future(self.listen())
         # Notify the main thread that we are ready to go.
         started_event.set()
         try:
             await self.listen_task
         finally:
             for c in list(self.connections):
                 c.close()
-            await asyncio.sleep(0.01, loop=self.loop)
+            await asyncio.sleep(0.01)
             if hasattr(self.loop, 'remove_reader'):
                 self.loop.remove_reader(self.sock.fileno())
             self.sock.close()
@@ -176,15 +176,15 @@ def close(self):
 
     async def handle(self):
         self.proxy_to_backend_task = asyncio.ensure_future(
-            self.proxy_to_backend(), loop=self.loop)
+            self.proxy_to_backend())
 
         self.proxy_from_backend_task = asyncio.ensure_future(
-            self.proxy_from_backend(), loop=self.loop)
+            self.proxy_from_backend())
 
         try:
             await asyncio.wait(
                 [self.proxy_to_backend_task, self.proxy_from_backend_task],
-                loop=self.loop, return_when=asyncio.FIRST_COMPLETED)
+                return_when=asyncio.FIRST_COMPLETED)
 
         finally:
             # Asyncio fails to properly remove the readers and writers
@@ -201,17 +201,14 @@ async def handle(self):
 
     async def _read(self, sock, n):
         read_task = asyncio.ensure_future(
-            self.loop.sock_recv(sock, n),
-            loop=self.loop)
+            self.loop.sock_recv(sock, n))
         conn_event_task = asyncio.ensure_future(
-            self.connectivity_loss.wait(),
-            loop=self.loop)
+            self.connectivity_loss.wait())
 
         try:
             await asyncio.wait(
                 [read_task, conn_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.connectivity_loss.is_set():
                 return None
@@ -225,15 +222,14 @@ async def _read(self, sock, n):
 
     async def _write(self, sock, data):
         write_task = asyncio.ensure_future(
-            self.loop.sock_sendall(sock, data), loop=self.loop)
+            self.loop.sock_sendall(sock, data))
         conn_event_task = asyncio.ensure_future(
-            self.connectivity_loss.wait(), loop=self.loop)
+            self.connectivity_loss.wait())
 
         try:
             await asyncio.wait(
                 [write_task, conn_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.connectivity_loss.is_set():
                 return None

asyncpg/connect_utils.py

@@ -531,7 +531,7 @@ async def _connect_addr(*, addr, loop, timeout, params, config,
     before = time.monotonic()
     try:
         tr, pr = await asyncio.wait_for(
-            connector, timeout=timeout, loop=loop)
+            connector, timeout=timeout)
     except asyncio.CancelledError:
         connector.add_done_callback(_close_leaked_connection)
         raise
@@ -540,7 +540,7 @@ async def _connect_addr(*, addr, loop, timeout, params, config,
     try:
         if timeout <= 0:
             raise asyncio.TimeoutError
-        await asyncio.wait_for(connected, loop=loop, timeout=timeout)
+        await asyncio.wait_for(connected, timeout=timeout)
     except (Exception, asyncio.CancelledError):
         tr.close()
         raise
@@ -581,7 +581,7 @@ async def _negotiate_ssl_connection(host, port, conn_factory, *, loop, ssl,
     # accept SSLRequests. If the SSLRequest is accepted but either the SSL
     # negotiation fails or the PostgreSQL user isn't permitted to use SSL,
     # there's nothing that would attempt to reconnect with a non-SSL socket.
-    reader, writer = await asyncio.open_connection(host, port, loop=loop)
+    reader, writer = await asyncio.open_connection(host, port)
 
     tr = writer.transport
     try:
@@ -632,18 +632,18 @@ async def _create_ssl_connection(protocol_factory, host, port, *,
 
 async def _open_connection(*, loop, addr, params: _ConnectionParameters):
     if isinstance(addr, str):
-        r, w = await asyncio.open_unix_connection(addr, loop=loop)
+        r, w = await asyncio.open_unix_connection(addr)
     else:
         if params.ssl:
             r, w = await _negotiate_ssl_connection(
                 *addr,
-                functools.partial(asyncio.open_connection, loop=loop),
+                asyncio.open_connection,
                 loop=loop,
                 ssl=params.ssl,
                 server_hostname=addr[0],
                 ssl_is_advisory=params.ssl_is_advisory)
         else:
-            r, w = await asyncio.open_connection(*addr, loop=loop)
+            r, w = await asyncio.open_connection(*addr)
             _set_nodelay(_get_socket(w.transport))
 
     return r, w

asyncpg/pool.py

@@ -199,7 +199,7 @@ async def release(self, timeout):
                 started = time.monotonic()
                 await asyncio.wait_for(
                     self._con._protocol._wait_for_cancellation(),
-                    budget, loop=self._pool._loop)
+                    budget)
                 if budget is not None:
                     budget -= time.monotonic() - started
 
@@ -362,7 +362,7 @@ def __init__(self, *connect_args,
         self._holders = []
         self._initialized = False
         self._initializing = False
-        self._queue = asyncio.LifoQueue(maxsize=self._maxsize, loop=self._loop)
+        self._queue = asyncio.LifoQueue(maxsize=self._maxsize)
 
         self._working_addr = None
         self._working_config = None
@@ -424,7 +424,7 @@ async def _initialize(self):
                         break
                     connect_tasks.append(ch.connect())
 
-                await asyncio.gather(*connect_tasks, loop=self._loop)
+                await asyncio.gather(*connect_tasks)
 
     def set_connect_args(self, dsn=None, **connect_kwargs):
         r"""Set the new connection arguments for this pool.
@@ -604,7 +604,7 @@ async def _acquire_impl():
             return await _acquire_impl()
         else:
             return await asyncio.wait_for(
-                _acquire_impl(), timeout=timeout, loop=self._loop)
+                _acquire_impl(), timeout=timeout)
 
     async def release(self, connection, *, timeout=None):
         """Release a database connection back to the pool.
@@ -642,7 +642,7 @@ async def release(self, connection, *, timeout=None):
         # Use asyncio.shield() to guarantee that task cancellation
         # does not prevent the connection from being returned to the
         # pool properly.
-        return await asyncio.shield(ch.release(timeout), loop=self._loop)
+        return await asyncio.shield(ch.release(timeout))
 
     async def close(self):
         """Attempt to gracefully close all connections in the pool.
@@ -673,11 +673,11 @@ async def close(self):
 
             release_coros = [
                 ch.wait_until_released() for ch in self._holders]
-            await asyncio.gather(*release_coros, loop=self._loop)
+            await asyncio.gather(*release_coros)
 
             close_coros = [
                 ch.close() for ch in self._holders]
-            await asyncio.gather(*close_coros, loop=self._loop)
+            await asyncio.gather(*close_coros)
 
         except (Exception, asyncio.CancelledError):
             self.terminate()

asyncpg/protocol/protocol.pyx

@@ -93,7 +93,7 @@ cdef class BaseProtocol(CoreProtocol):
 
         self.closing = False
         self.is_reading = True
-        self.writing_allowed = asyncio.Event(loop=self.loop)
+        self.writing_allowed = asyncio.Event()
         self.writing_allowed.set()
 
         self.timeout_handle = None
@@ -348,8 +348,7 @@ cdef class BaseProtocol(CoreProtocol):
                         with timer:
                             await asyncio.wait_for(
                                 sink(buffer),
-                                timeout=timer.get_remaining_budget(),
-                                loop=self.loop)
+                                timeout=timer.get_remaining_budget())
                     except (Exception, asyncio.CancelledError) as ex:
                         # Abort the COPY operation on any error in
                         # output sink.
@@ -456,8 +455,7 @@ cdef class BaseProtocol(CoreProtocol):
                         with timer:
                             chunk = await asyncio.wait_for(
                                 iterator.__anext__(),
-                                timeout=timer.get_remaining_budget(),
-                                loop=self.loop)
+                                timeout=timer.get_remaining_budget())
                         self._write_copy_data_msg(chunk)
                 except builtins.StopAsyncIteration:
                     pass

tests/test_adversity.py

@@ -62,6 +62,6 @@ def kill_connectivity():
                 workers = [worker(pool) for _ in range(concurrency)]
                 self.loop.call_later(1, kill_connectivity)
                 await asyncio.gather(
-                    *workers, loop=self.loop, return_exceptions=True)
+                    *workers, return_exceptions=True)
             finally:
                 pool.terminate()

tests/test_cancellation.py

@@ -55,7 +55,7 @@ async def test8():
             with self.subTest(testfunc=test), self.assertRunUnder(1):
                 st = await self.con.prepare('SELECT pg_sleep(20)')
                 task = self.loop.create_task(st.fetch())
-                await asyncio.sleep(0.05, loop=self.loop)
+                await asyncio.sleep(0.05)
                 task.cancel()
 
                 with self.assertRaises(asyncio.CancelledError):
@@ -67,7 +67,7 @@ async def test8():
     async def test_cancellation_02(self):
         st = await self.con.prepare('SELECT 1')
         task = self.loop.create_task(st.fetch())
-        await asyncio.sleep(0.05, loop=self.loop)
+        await asyncio.sleep(0.05)
         task.cancel()
         self.assertEqual(await task, [(1,)])
 
@@ -76,7 +76,7 @@ async def test_cancellation_03(self):
             async with self.con.transaction():
                 task = self.loop.create_task(
                     self.con.fetch('SELECT pg_sleep(20)'))
-                await asyncio.sleep(0.05, loop=self.loop)
+                await asyncio.sleep(0.05)
                 task.cancel()
 
                 with self.assertRaises(asyncio.CancelledError):
@@ -90,7 +90,7 @@ async def test_cancellation_03(self):
 
     async def test_cancellation_04(self):
         await self.con.fetchval('SELECT pg_sleep(0)')
-        waiter = asyncio.Future(loop=self.loop)
+        waiter = asyncio.Future()
         self.con._cancel_current_command(waiter)
         await waiter
         self.assertEqual(await self.con.fetchval('SELECT 42'), 42)

tests/test_connect.py

@@ -934,14 +934,14 @@ def test_connect_pgpass_inaccessible_directory(self):
     async def test_connect_args_validation(self):
         for val in {-1, 'a', True, False, 0}:
             with self.assertRaisesRegex(ValueError, 'greater than 0'):
-                await asyncpg.connect(command_timeout=val, loop=self.loop)
+                await asyncpg.connect(command_timeout=val)
 
         for arg in {'max_cacheable_statement_size',
                     'max_cached_statement_lifetime',
                     'statement_cache_size'}:
             for val in {None, -1, True, False}:
                 with self.assertRaisesRegex(ValueError, 'greater or equal'):
-                    await asyncpg.connect(**{arg: val}, loop=self.loop)
+                    await asyncpg.connect(**{arg: val})
 
 
 class TestConnection(tb.ConnectedTestCase):
@@ -1040,8 +1040,7 @@ async def test_connection_implicit_host(self):
         con = await asyncpg.connect(
             port=conn_spec.get('port'),
             database=conn_spec.get('database'),
-            user=conn_spec.get('user'),
-            loop=self.loop)
+            user=conn_spec.get('user'))
         await con.close()
 
 
@@ -1207,7 +1206,7 @@ async def worker():
                 self.assertEqual(await con.fetchval('SELECT 43'), 43)
 
         tasks = [worker() for _ in range(100)]
-        await asyncio.gather(*tasks, loop=self.loop)
+        await asyncio.gather(*tasks)
         await pool.close()