gh-90908: Document asyncio.Task.cancelling() and asyncio.Task.uncancel()

Doc/library/asyncio-task.rst

@@ -293,6 +293,8 @@ It is recommended that coroutines use ``try/finally`` blocks to robustly
 perform clean-up logic. In case :exc:`asyncio.CancelledError`
 is explicitly caught, it should generally be propagated when
 clean-up is complete. Most code can safely ignore :exc:`asyncio.CancelledError`.
+If a task needs to continue despite receiving an :exc:`asyncio.CancelledError`,
+it should :func:`uncancel itself <asyncio.Task.uncancel>`.
 
 Important asyncio components, like :class:`asyncio.TaskGroup` and the
 :func:`asyncio.timeout` context manager, are implemented using cancellation
@@ -1064,6 +1066,36 @@ Task Object
       :meth:`cancel` and the wrapped coroutine propagated the
       :exc:`CancelledError` exception thrown into it.
 
+   .. method:: cancelling()
+
+      Return the number of cancellation requests to this Task, i.e.,
+      the number of calls to :meth:`cancel`.
+
+      Note that if this number is greater than zero but the Task is
+      still executing, :meth:`cancelled` will still return ``False``.
+      It's because this number can be lowered by calling :meth:`uncancel`,
+      which can lead to the task not being cancelled after all if the
+      cancellation requests go down to zero.
+
+   .. method:: uncancel()
+
+      Decrement the count of cancellation requests to this Task.
+
+      Returns the remaining number of cancellation requests.
+
+      This should be used by tasks that catch :exc:`CancelledError`
+      and wish to continue indefinitely until they are cancelled again::
+
+        async def resilient_task():
+            try:
+                await do_work()
+            except asyncio.CancelledError:
+                asyncio.current_task().uncancel()
+                await do_work()
+
+      Note that once execution of a cancelled task completed, further
+      calls to :meth:`uncancel` are ineffective.
+
    .. method:: done()
 
       Return ``True`` if the Task is *done*.

Lib/test/test_asyncio/test_tasks.py

@@ -534,14 +534,32 @@ async def task():
         try:
             t = self.new_task(loop, task())
             loop.run_until_complete(asyncio.sleep(0.01))
-            self.assertTrue(t.cancel())  # Cancel first sleep
+
+            # Cancel first sleep
+            self.assertTrue(t.cancel())
             self.assertIn(" cancelling ", repr(t))
+            self.assertEqual(t.cancelling(), 1)
+            self.assertFalse(t.cancelled())  # Task is still not complete
             loop.run_until_complete(asyncio.sleep(0.01))
-            self.assertNotIn(" cancelling ", repr(t))  # after .uncancel()
-            self.assertTrue(t.cancel())  # Cancel second sleep
 
+            # after .uncancel()
+            self.assertNotIn(" cancelling ", repr(t))
+            self.assertEqual(t.cancelling(), 0)
+            self.assertFalse(t.cancelled())  # Task is still not complete
+
+            # Cancel second sleep
+            self.assertTrue(t.cancel())
+            self.assertEqual(t.cancelling(), 1)
+            self.assertFalse(t.cancelled())  # Task is still not complete
             with self.assertRaises(asyncio.CancelledError):
                 loop.run_until_complete(t)
+            self.assertTrue(t.cancelled())  # Finally, task complete
+            self.assertTrue(t.done())
+
+            # uncancel is no longer effective after the task is complete
+            t.uncancel()
+            self.assertTrue(t.cancelled())
+            self.assertTrue(t.done())
         finally:
             loop.close()