bpo-42392: Remove loop parameter from asyncio.tasks and asyncio.subprocess (GH-23521)

Lib/asyncio/base_events.py

@@ -350,7 +350,7 @@ async def start_serving(self):
         self._start_serving()
         # Skip one loop iteration so that all 'loop.add_reader'
         # go through.
-        await tasks.sleep(0, loop=self._loop)
+        await tasks.sleep(0)
 
     async def serve_forever(self):
         if self._serving_forever_fut is not None:
@@ -541,8 +541,7 @@ async def shutdown_asyncgens(self):
 
         results = await tasks.gather(
             *[ag.aclose() for ag in closing_agens],
-            return_exceptions=True,
+            return_exceptions=True)
-            loop=self)
 
         for result, agen in zip(results, closing_agens):
             if isinstance(result, Exception):
@@ -1457,7 +1456,7 @@ async def create_server(
             fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                   flags=flags)
                   for host in hosts]
-            infos = await tasks.gather(*fs, loop=self)
+            infos = await tasks.gather(*fs)
             infos = set(itertools.chain.from_iterable(infos))
 
             completed = False
@@ -1515,7 +1514,7 @@ async def create_server(
             server._start_serving()
             # Skip one loop iteration so that all 'loop.add_reader'
             # go through.
-            await tasks.sleep(0, loop=self)
+            await tasks.sleep(0)
 
         if self._debug:
             logger.info("%r is serving", server)

Lib/asyncio/runners.py

@@ -60,8 +60,7 @@ def _cancel_all_tasks(loop):
     for task in to_cancel:
         task.cancel()
 
-    loop.run_until_complete(
+    loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True))
-        tasks.gather(*to_cancel, loop=loop, return_exceptions=True))
 
     for task in to_cancel:
         if task.cancelled():

Lib/asyncio/subprocess.py

@@ -1,7 +1,6 @@
 __all__ = 'create_subprocess_exec', 'create_subprocess_shell'
 
 import subprocess
-import warnings
 
 from . import events
 from . import protocols
@@ -193,24 +192,14 @@ async def communicate(self, input=None):
             stderr = self._read_stream(2)
         else:
             stderr = self._noop()
-        stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr,
+        stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr)
-                                                   loop=self._loop)
         await self.wait()
         return (stdout, stderr)
 
 
 async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
-                                  loop=None, limit=streams._DEFAULT_LIMIT,
+                                  limit=streams._DEFAULT_LIMIT, **kwds):
-                                  **kwds):
+    loop = events.get_running_loop()
-    if loop is None:
-        loop = events.get_event_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8 "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning,
-                      stacklevel=2
-        )
-
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
                                                         loop=loop)
     transport, protocol = await loop.subprocess_shell(
@@ -221,16 +210,9 @@ async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
 
 
 async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
-                                 stderr=None, loop=None,
+                                 stderr=None, limit=streams._DEFAULT_LIMIT,
-                                 limit=streams._DEFAULT_LIMIT, **kwds):
+                                 **kwds):
-    if loop is None:
+    loop = events.get_running_loop()
-        loop = events.get_event_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8 "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning,
-                      stacklevel=2
-        )
     protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
                                                         loop=loop)
     transport, protocol = await loop.subprocess_exec(

Lib/asyncio/tasks.py

@@ -370,7 +370,7 @@ def create_task(coro, *, name=None):
 ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
 
 
-async def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
+async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
     """Wait for the Futures and coroutines given by fs to complete.
 
     The fs iterable must not be empty.
@@ -393,12 +393,7 @@ async def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
     if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
         raise ValueError(f'Invalid return_when value: {return_when}')
 
-    if loop is None:
+    loop = events.get_running_loop()
-        loop = events.get_running_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
 
     fs = set(fs)
 
@@ -418,7 +413,7 @@ def _release_waiter(waiter, *args):
         waiter.set_result(None)
 
 
-async def wait_for(fut, timeout, *, loop=None):
+async def wait_for(fut, timeout):
     """Wait for the single Future or coroutine to complete, with timeout.
 
     Coroutine will be wrapped in Task.
@@ -431,12 +426,7 @@ async def wait_for(fut, timeout, *, loop=None):
 
     This function is a coroutine.
     """
-    if loop is None:
+    loop = events.get_running_loop()
-        loop = events.get_running_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
 
     if timeout is None:
         return await fut
@@ -556,7 +546,7 @@ async def _cancel_and_wait(fut, loop):
 
 
 # This is *not* a @coroutine!  It is just an iterator (yielding Futures).
-def as_completed(fs, *, loop=None, timeout=None):
+def as_completed(fs, *, timeout=None):
     """Return an iterator whose values are coroutines.
 
     When waiting for the yielded coroutines you'll get the results (or
@@ -580,12 +570,7 @@ def as_completed(fs, *, loop=None, timeout=None):
     from .queues import Queue  # Import here to avoid circular import problem.
     done = Queue()
 
-    if loop is None:
+    loop = events.get_event_loop()
-        loop = events.get_event_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
     todo = {ensure_future(f, loop=loop) for f in set(fs)}
     timeout_handle = None
 
@@ -630,19 +615,13 @@ def __sleep0():
     yield
 
 
-async def sleep(delay, result=None, *, loop=None):
+async def sleep(delay, result=None):
     """Coroutine that completes after a given time (in seconds)."""
     if delay <= 0:
         await __sleep0()
         return result
 
-    if loop is None:
+    loop = events.get_running_loop()
-        loop = events.get_running_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
-
     future = loop.create_future()
     h = loop.call_later(delay,
                         futures._set_result_unless_cancelled,
@@ -717,7 +696,7 @@ def cancel(self, msg=None):
         return ret
 
 
-def gather(*coros_or_futures, loop=None, return_exceptions=False):
+def gather(*coros_or_futures, return_exceptions=False):
     """Return a future aggregating results from the given coroutines/futures.
 
     Coroutines will be wrapped in a future and scheduled in the event
@@ -748,12 +727,7 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
     gather won't cancel any other awaitables.
     """
     if not coros_or_futures:
-        if loop is None:
+        loop = events.get_event_loop()
-            loop = events.get_event_loop()
-        else:
-            warnings.warn("The loop argument is deprecated since Python 3.8, "
-                          "and scheduled for removal in Python 3.10.",
-                          DeprecationWarning, stacklevel=2)
         outer = loop.create_future()
         outer.set_result([])
         return outer
@@ -817,6 +791,7 @@ def _done_callback(fut):
     children = []
     nfuts = 0
     nfinished = 0
+    loop = None
     for arg in coros_or_futures:
         if arg not in arg_to_fut:
             fut = ensure_future(arg, loop=loop)
@@ -843,7 +818,7 @@ def _done_callback(fut):
     return outer
 
 
-def shield(arg, *, loop=None):
+def shield(arg):
     """Wait for a future, shielding it from cancellation.
 
     The statement
@@ -869,11 +844,7 @@ def shield(arg, *, loop=None):
         except CancelledError:
             res = None
     """
-    if loop is not None:
+    inner = ensure_future(arg)
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
-    inner = ensure_future(arg, loop=loop)
     if inner.done():
         # Shortcut.
         return inner

Lib/asyncio/unix_events.py

@@ -323,7 +323,7 @@ async def create_unix_server(
             server._start_serving()
             # Skip one loop iteration so that all 'loop.add_reader'
             # go through.
-            await tasks.sleep(0, loop=self)
+            await tasks.sleep(0)
 
         return server
 

Lib/test/test_asyncio/test_queues.py

@@ -255,6 +255,7 @@ def test_get_with_waiting_putters(self):
 
     def test_why_are_getters_waiting(self):
         # From issue #268.
+        asyncio.set_event_loop(self.loop)
 
         async def consumer(queue, num_expected):
             for _ in range(num_expected):
@@ -276,8 +277,7 @@ async def create_queue():
 
         self.loop.run_until_complete(
             asyncio.gather(producer(q, producer_num_items),
-                           consumer(q, producer_num_items),
+                           consumer(q, producer_num_items)),
-                           loop=self.loop),
             )
 
     def test_cancelled_getters_not_being_held_in_self_getters(self):
@@ -498,6 +498,7 @@ def test_put_with_waiting_getters(self):
 
     def test_why_are_putters_waiting(self):
         # From issue #265.
+        asyncio.set_event_loop(self.loop)
 
         async def create_queue():
             q = asyncio.Queue(2)
@@ -519,8 +520,7 @@ async def getter():
         t1 = putter(1)
         t2 = putter(2)
         t3 = putter(3)
-        self.loop.run_until_complete(
+        self.loop.run_until_complete(asyncio.gather(getter(), t0, t1, t2, t3))
-            asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop))
 
     def test_cancelled_puts_not_being_held_in_self_putters(self):
         def a_generator():

Lib/test/test_asyncio/test_subprocess.py

@@ -635,26 +635,6 @@ async def execute():
 
         self.assertIsNone(self.loop.run_until_complete(execute()))
 
-    def test_exec_loop_deprecated(self):
-        async def go():
-            with self.assertWarns(DeprecationWarning):
-                proc = await asyncio.create_subprocess_exec(
-                    sys.executable, '-c', 'pass',
-                    loop=self.loop,
-                )
-            await proc.wait()
-        self.loop.run_until_complete(go())
-
-    def test_shell_loop_deprecated(self):
-        async def go():
-            with self.assertWarns(DeprecationWarning):
-                proc = await asyncio.create_subprocess_shell(
-                    "exit 0",
-                    loop=self.loop,
-                )
-            await proc.wait()
-        self.loop.run_until_complete(go())
-
 
 if sys.platform != 'win32':
     # Unix

Lib/test/test_asyncio/test_tasks.py

@@ -155,7 +155,7 @@ async def run():
         self.loop.run_until_complete(
             asyncio.gather(*[
                 self.new_task(self.loop, run()) for _ in range(100)
-            ], loop=self.loop))
+            ]))
 
     def test_other_loop_future(self):
         other_loop = asyncio.new_event_loop()
@@ -1603,19 +1603,18 @@ def sleeper(dt, x):
 
         async def foo():
             values = []
-            for f in asyncio.as_completed([b, c, a], loop=loop):
+            for f in asyncio.as_completed([b, c, a]):
                 values.append(await f)
             return values
-        with self.assertWarns(DeprecationWarning):
+
-            res = loop.run_until_complete(self.new_task(loop, foo()))
+        res = loop.run_until_complete(self.new_task(loop, foo()))
         self.assertAlmostEqual(0.15, loop.time())
         self.assertTrue('a' in res[:2])
         self.assertTrue('b' in res[:2])
         self.assertEqual(res[2], 'c')
 
         # Doing it again should take no time and exercise a different path.
-        with self.assertWarns(DeprecationWarning):
+        res = loop.run_until_complete(self.new_task(loop, foo()))
-            res = loop.run_until_complete(self.new_task(loop, foo()))
         self.assertAlmostEqual(0.15, loop.time())
 
     def test_as_completed_with_timeout(self):
@@ -1633,7 +1632,7 @@ def gen():
 
         async def foo():
             values = []
-            for f in asyncio.as_completed([a, b], timeout=0.12, loop=loop):
+            for f in asyncio.as_completed([a, b], timeout=0.12):
                 if values:
                     loop.advance_time(0.02)
                 try:
@@ -1643,8 +1642,7 @@ async def foo():
                     values.append((2, exc))
             return values
 
-        with self.assertWarns(DeprecationWarning):
+        res = loop.run_until_complete(self.new_task(loop, foo()))
-            res = loop.run_until_complete(self.new_task(loop, foo()))
         self.assertEqual(len(res), 2, res)
         self.assertEqual(res[0], (1, 'a'))
         self.assertEqual(res[1][0], 2)
@@ -1667,12 +1665,11 @@ def gen():
         a = asyncio.sleep(0.01, 'a')
 
         async def foo():
-            for f in asyncio.as_completed([a], timeout=1, loop=loop):
+            for f in asyncio.as_completed([a], timeout=1):
                 v = await f
                 self.assertEqual(v, 'a')
 
-        with self.assertWarns(DeprecationWarning):
+        loop.run_until_complete(self.new_task(loop, foo()))
-            loop.run_until_complete(self.new_task(loop, foo()))
 
     def test_as_completed_reverse_wait(self):
 
@@ -1682,13 +1679,13 @@ def gen():
             yield 0
 
         loop = self.new_test_loop(gen)
+        asyncio.set_event_loop(loop)
 
         a = asyncio.sleep(0.05, 'a')
         b = asyncio.sleep(0.10, 'b')
         fs = {a, b}
 
-        with self.assertWarns(DeprecationWarning):
+        futs = list(asyncio.as_completed(fs))
-            futs = list(asyncio.as_completed(fs, loop=loop))
         self.assertEqual(len(futs), 2)
 
         x = loop.run_until_complete(futs[1])
@@ -1709,12 +1706,13 @@ def gen():
             yield 0.05
 
         loop = self.new_test_loop(gen)
+        asyncio.set_event_loop(loop)
 
         a = asyncio.sleep(0.05, 'a')
         b = asyncio.sleep(0.05, 'b')
         fs = {a, b}
-        with self.assertWarns(DeprecationWarning):
+
-            futs = list(asyncio.as_completed(fs, loop=loop))
+        futs = list(asyncio.as_completed(fs))
         self.assertEqual(len(futs), 2)
         waiter = asyncio.wait(futs)
         # Deprecation from passing coros in futs to asyncio.wait()
@@ -1734,14 +1732,12 @@ def coro(s):
             def runner():
                 result = []
                 c = coro('ham')
-                for f in asyncio.as_completed([c, c, coro('spam')],
+                for f in asyncio.as_completed([c, c, coro('spam')]):
-                                              loop=self.loop):
                     result.append((yield from f))
                 return result
 
-        with self.assertWarns(DeprecationWarning):
+        fut = self.new_task(self.loop, runner())
-            fut = self.new_task(self.loop, runner())
+        self.loop.run_until_complete(fut)
-            self.loop.run_until_complete(fut)
         result = fut.result()
         self.assertEqual(set(result), {'ham', 'spam'})
         self.assertEqual(len(result), 2)
@@ -2018,7 +2014,7 @@ def test_current_task(self):
         self.assertIsNone(asyncio.current_task(loop=self.loop))
 
         async def coro(loop):
-            self.assertIs(asyncio.current_task(loop=loop), task)
+            self.assertIs(asyncio.current_task(), task)
 
             self.assertIs(asyncio.current_task(None), task)
             self.assertIs(asyncio.current_task(), task)
@@ -2034,16 +2030,16 @@ def test_current_task_with_interleaving_tasks(self):
         fut2 = self.new_future(self.loop)
 
         async def coro1(loop):
-            self.assertTrue(asyncio.current_task(loop=loop) is task1)
+            self.assertTrue(asyncio.current_task() is task1)
             await fut1
-            self.assertTrue(asyncio.current_task(loop=loop) is task1)
+            self.assertTrue(asyncio.current_task() is task1)
             fut2.set_result(True)
 
         async def coro2(loop):
-            self.assertTrue(asyncio.current_task(loop=loop) is task2)
+            self.assertTrue(asyncio.current_task() is task2)
             fut1.set_result(True)
             await fut2
-            self.assertTrue(asyncio.current_task(loop=loop) is task2)
+            self.assertTrue(asyncio.current_task() is task2)
 
         task1 = self.new_task(self.loop, coro1(self.loop))
         task2 = self.new_task(self.loop, coro2(self.loop))
@@ -2210,10 +2206,10 @@ def test_as_completed_invalid_args(self):
 
         # as_completed() expects a list of futures, not a future instance
         self.assertRaises(TypeError, self.loop.run_until_complete,
-            asyncio.as_completed(fut, loop=self.loop))
+            asyncio.as_completed(fut))
         coro = coroutine_function()
         self.assertRaises(TypeError, self.loop.run_until_complete,
-            asyncio.as_completed(coro, loop=self.loop))
+            asyncio.as_completed(coro))
         coro.close()
 
     def test_wait_invalid_args(self):
@@ -2511,6 +2507,7 @@ def test_cancel_gather_1(self):
         """Ensure that a gathering future refuses to be cancelled once all
         children are done"""
         loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
         self.addCleanup(loop.close)
 
         fut = self.new_future(loop)
@@ -2518,7 +2515,7 @@ def test_cancel_gather_1(self):
         # gathering task is done at the same time as the child future
         def child_coro():
             return (yield from fut)
-        gather_future = asyncio.gather(child_coro(), loop=loop)
+        gather_future = asyncio.gather(child_coro())
         gather_task = asyncio.ensure_future(gather_future, loop=loop)
 
         cancel_result = None
@@ -2555,8 +2552,7 @@ async def test():
                     while True:
                         time += 0.05
                         await asyncio.gather(asyncio.sleep(0.05),
-                                             return_exceptions=True,
+                                             return_exceptions=True)
-                                             loop=loop)
                         if time > 1:
                             return
 
@@ -2771,7 +2767,7 @@ async def main():
                 task = loop.create_task(sub(random.randint(0, 10)))
                 tasks.append(task)
 
-            await asyncio.gather(*tasks, loop=loop)
+            await asyncio.gather(*tasks)
 
         loop = asyncio.new_event_loop()
         try:
@@ -3328,9 +3324,6 @@ def _check_empty_sequence(self, seq_or_iter):
         self._run_loop(self.one_loop)
         self.assertTrue(fut.done())
         self.assertEqual(fut.result(), [])
-        with self.assertWarns(DeprecationWarning):
-            fut = asyncio.gather(*seq_or_iter, loop=self.other_loop)
-        self.assertIs(fut._loop, self.other_loop)
 
     def test_constructor_empty_sequence(self):
         self._check_empty_sequence([])
@@ -3343,8 +3336,6 @@ def test_constructor_heterogenous_futures(self):
         fut2 = self.other_loop.create_future()
         with self.assertRaises(ValueError):
             asyncio.gather(fut1, fut2)
-        with self.assertRaises(ValueError):
-            asyncio.gather(fut1, loop=self.other_loop)
 
     def test_constructor_homogenous_futures(self):
         children = [self.other_loop.create_future() for i in range(3)]
@@ -3352,7 +3343,7 @@ def test_constructor_homogenous_futures(self):
         self.assertIs(fut._loop, self.other_loop)
         self._run_loop(self.other_loop)
         self.assertFalse(fut.done())
-        fut = asyncio.gather(*children, loop=self.other_loop)
+        fut = asyncio.gather(*children)
         self.assertIs(fut._loop, self.other_loop)
         self._run_loop(self.other_loop)
         self.assertFalse(fut.done())
@@ -3423,9 +3414,10 @@ async def coro():
         self.one_loop.run_until_complete(fut)
 
         self.set_event_loop(self.other_loop, cleanup=False)
+        asyncio.set_event_loop(self.other_loop)
         gen3 = coro()
         gen4 = coro()
-        fut2 = asyncio.gather(gen3, gen4, loop=self.other_loop)
+        fut2 = asyncio.gather(gen3, gen4)
         self.assertIs(fut2._loop, self.other_loop)
         self.other_loop.run_until_complete(fut2)
 
@@ -3435,7 +3427,7 @@ def test_duplicate_coroutines(self):
             def coro(s):
                 return s
         c = coro('abc')
-        fut = asyncio.gather(c, c, coro('def'), c, loop=self.one_loop)
+        fut = asyncio.gather(c, c, coro('def'), c)
         self._run_loop(self.one_loop)
         self.assertEqual(fut.result(), ['abc', 'abc', 'def', 'abc'])
 
@@ -3455,7 +3447,7 @@ async def inner():
 
         async def outer():
             nonlocal proof, gatherer
-            gatherer = asyncio.gather(child1, child2, loop=self.one_loop)
+            gatherer = asyncio.gather(child1, child2)
             await gatherer
             proof += 100
 
@@ -3482,7 +3474,7 @@ async def inner(f):
         b = self.one_loop.create_future()
 
         async def outer():
-            await asyncio.gather(inner(a), inner(b), loop=self.one_loop)
+            await asyncio.gather(inner(a), inner(b))
 
         f = asyncio.ensure_future(outer(), loop=self.one_loop)
         test_utils.run_briefly(self.one_loop)
@@ -3621,11 +3613,6 @@ async def coro():
         self.loop.run_until_complete(coro())
         self.assertEqual(result, 11)
 
-    def test_loop_argument_is_deprecated(self):
-        # Remove test when loop argument is removed in Python 3.10
-        with self.assertWarns(DeprecationWarning):
-            self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
-
 
 class WaitTests(test_utils.TestCase):
     def setUp(self):
@@ -3638,18 +3625,6 @@ def tearDown(self):
         self.loop = None
         super().tearDown()
 
-    def test_loop_argument_is_deprecated_in_wait(self):
-        # Remove test when loop argument is removed in Python 3.10
-        with self.assertWarns(DeprecationWarning):
-            self.loop.run_until_complete(
-                asyncio.wait([coroutine_function()], loop=self.loop))
-
-    def test_loop_argument_is_deprecated_in_wait_for(self):
-        # Remove test when loop argument is removed in Python 3.10
-        with self.assertWarns(DeprecationWarning):
-            self.loop.run_until_complete(
-                asyncio.wait_for(coroutine_function(), 0.01, loop=self.loop))
-
     def test_coro_is_deprecated_in_wait(self):
         # Remove test when passing coros to asyncio.wait() is removed in 3.11
         with self.assertWarns(DeprecationWarning):
@@ -3701,7 +3676,7 @@ def coro2():
                 return 'ok2'
 
         async def inner():
-            return await asyncio.gather(coro1(), coro2(), loop=self.loop)
+            return await asyncio.gather(coro1(), coro2())
 
         result = self.loop.run_until_complete(inner())
         self.assertEqual(['ok1', 'ok2'], result)

Misc/NEWS.d/next/Library/2020-11-26-12-40-16.bpo-42392.GbmdHE.rst

@@ -0,0 +1,2 @@
+Remove loop parameter from ``asyncio.subprocess`` and ``asyncio.tasks``
+functions. Patch provided by Yurii Karabas.