gh-121468: Show asyncio information in pdb

Doc/library/pdb.rst

@@ -304,16 +304,20 @@ sets a global variable ``$foo`` which you can use in the debugger session.  The
 less likely to interfere with your program compared to using normal variables
 like ``foo = 1``.
 
-There are three preset *convenience variables*:
+There are four preset *convenience variables*:
 
 * ``$_frame``: the current frame you are debugging
 * ``$_retval``: the return value if the frame is returning
 * ``$_exception``: the exception if the frame is raising an exception
+* ``$_asynctask``: the asyncio task if pdb stops in an async function
 
 .. versionadded:: 3.12
 
    Added the *convenience variable* feature.
 
+.. versionadded:: 3.14
+   Added the ``$_asynctask`` convenience variable.
+
 .. index::
    pair: .pdbrc; file
    triple: debugger; configuration; file

Doc/whatsnew/3.14.rst

@@ -811,6 +811,9 @@ pdb
   fill in a 4-space indentation now, instead of inserting a ``\t`` character.
   (Contributed by Tian Gao in :gh:`130471`.)
 
+* ``$_asynctask`` is added to access the current asyncio task if applicable.
+  (Contributed by Tian Gao in :gh:`124367`.)
+
 
 pickle
 ------

Lib/pdb.py

@@ -79,6 +79,7 @@
 import codeop
 import pprint
 import signal
+import asyncio
 import inspect
 import textwrap
 import tokenize
@@ -363,6 +364,8 @@ def __init__(self, completekey='tab', stdin=None, stdout=None, skip=None,
         self._chained_exceptions = tuple()
         self._chained_exception_index = 0
 
+        self._current_task = None
+
     def set_trace(self, frame=None, *, commands=None):
         Pdb._last_pdb_instance = self
         if frame is None:
@@ -405,7 +408,8 @@ def setup(self, f, tb):
             tb = tb.tb_next
         self.curframe = self.stack[self.curindex][0]
         self.set_convenience_variable(self.curframe, '_frame', self.curframe)
-
+        if self._current_task:
+            self.set_convenience_variable(self.curframe, '_asynctask', self._current_task)
         self._save_initial_file_mtime(self.curframe)
 
         if self._chained_exceptions:
@@ -616,6 +620,13 @@ def _hold_exceptions(self, exceptions):
             self._chained_exceptions = tuple()
             self._chained_exception_index = 0
 
+    def _get_asyncio_task(self):
+        try:
+            task = asyncio.current_task()
+        except RuntimeError:
+            task = None
+        return task
+
     def interaction(self, frame, tb_or_exc):
         # Restore the previous signal handler at the Pdb prompt.
         if Pdb._previous_sigint_handler:
@@ -626,6 +637,8 @@ def interaction(self, frame, tb_or_exc):
             else:
                 Pdb._previous_sigint_handler = None
 
+        self._current_task = self._get_asyncio_task()
+
         _chained_exceptions, tb = self._get_tb_and_exceptions(tb_or_exc)
         if isinstance(tb_or_exc, BaseException):
             assert tb is not None, "main exception must have a traceback"

Lib/test/test_pdb.py

@@ -16,7 +16,7 @@
 from contextlib import ExitStack, redirect_stdout
 from io import StringIO
 from test import support
-from test.support import force_not_colorized, os_helper
+from test.support import force_not_colorized, has_socket_support, os_helper
 from test.support.import_helper import import_module
 from test.support.pty_helper import run_pty, FakeInput
 from test.support.script_helper import kill_python
@@ -2059,6 +2059,30 @@ def test_pdb_next_command_for_generator():
     """
 
 if not SKIP_CORO_TESTS:
+    if has_socket_support:
+        def test_pdb_asynctask():
+            """Testing $_asynctask is accessible in async context
+
+            >>> import asyncio
+
+            >>> async def test():
+            ...     import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+
+            >>> def test_function():
+            ...     asyncio.run(test())
+
+            >>> with PdbTestInput([  # doctest: +ELLIPSIS
+            ...     '$_asynctask',
+            ...     'continue',
+            ... ]):
+            ...     test_function()
+            > <doctest test.test_pdb.test_pdb_asynctask[1]>(2)test()
+            -> import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+            (Pdb) $_asynctask
+            <Task pending name='Task-1' coro=<test() running at <doctest test.test_pdb.test_pdb_asynctask[1]>:2> ...
+            (Pdb) continue
+            """
+
     def test_pdb_next_command_for_coroutine():
         """Testing skip unwinding stack on yield for coroutines for "next" command
 

Misc/NEWS.d/next/Library/2025-03-11-21-52-33.gh-issue-121468.WsEP02.rst

@@ -0,0 +1,2 @@
+``$_asynctask`` is added as a :mod:`pdb` convenience variable to
+access the current asyncio task if applicable.