Apr 29, 2025 · Apr 13, 2025 · Apr 16, 2025 · Apr 16, 2025 · Apr 16, 2025 · Apr 16, 2025
diff --git a/Doc/library/pdb.rst b/Doc/library/pdb.rst
   .. versionadded:: 3.14
      The *commands* argument.


 .. awaitablefunction:: set_trace_async(*, header=None, commands=None)

   async version of :func:`set_trace`. This function should be used inside an
   async function with :keyword:`await`.

   .. code-block:: python

      async def f():
          await pdb.set_trace_async()

   :keyword:`await` statements are supported if the debugger is invoked by this function.

   .. versionadded:: 3.14

 .. function:: post_mortem(t=None)

   Enter post-mortem debugging of the given exception or
diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst
  backend by default, which is configurable.
  (Contributed by Tian Gao in :gh:`124533`.)

 * :func:`pdb.set_trace_async` is added to support debugging asyncio
  coroutines. :keyword:`await` statements are supported with this
  function.
  (Contributed by Tian Gao in :gh:`132576`.)


 pickle
 ------
diff --git a/Lib/pdb.py b/Lib/pdb.py
        self.commands_bnum = None # The breakpoint number for which we are
                                  # defining a list

        self.async_shim_frame = None
        self.async_awaitable = None

        self._chained_exceptions = tuple()
        self._chained_exception_index = 0


        super().set_trace(frame)

    async def set_trace_async(self, frame=None, *, commands=None):
        if self.async_awaitable is not None:
            # We are already in a set_trace_async call, do not mess with it
            return

        if frame is None:
            frame = sys._getframe().f_back

        # We need set_trace to set up the basics, however, this will call
        # set_stepinstr() will we need to compensate for, because we don't
        # want to trigger on calls
        self.set_trace(frame, commands=commands)
        # Changing the stopframe will disable trace dispatch on calls
        self.stopframe = frame
        # We need to stop tracing because we don't have the privilege to avoid
        # triggering tracing functions as normal, as we are not already in
        # tracing functions
        self.stop_trace()

        self.async_shim_frame = sys._getframe()
        self.async_awaitable = None

        while True:
            self.async_awaitable = None
            # Simulate a trace event
            # This should bring up pdb and make pdb believe it's debugging the
            # caller frame
            self.trace_dispatch(frame, "opcode", None)
            if self.async_awaitable is not None:
                try:
                    if self.breaks:
                        with self.set_enterframe(frame):
                            # set_continue requires enterframe to work
                            self.set_continue()
                        self.start_trace()
                    await self.async_awaitable
                except Exception:
                    self._error_exc()
            else:
                break

        self.async_shim_frame = None

        # start the trace (the actual command is already set by set_* calls)
        if self.returnframe is None and self.stoplineno == -1 and not self.breaks:
            # This means we did a continue without any breakpoints, we should not
            # start the trace
            return

        self.start_trace()

    def sigint_handler(self, signum, frame):
        if self.allow_kbdint:
            raise KeyboardInterrupt

        return True

    def default(self, line):
        if line[:1] == '!': line = line[1:].strip()
        locals = self.curframe.f_locals
        globals = self.curframe.f_globals
    def _exec_await(self, source, globals, locals):
        """ Run source code that contains await by playing with async shim frame"""
        # Put the source in an async function
        source_async = (
            "async def __pdb_await():\n" +
            textwrap.indent(source, "    ") + '\n' +
            "    __pdb_locals.update(locals())"
        )
        ns = globals | locals
        # We use __pdb_locals to do write back
        ns["__pdb_locals"] = locals
        exec(source_async, ns)
        self.async_awaitable = ns["__pdb_await"]()

    def _read_code(self, line):
        buffer = line
        is_await_code = False
        code = None
        try:
            buffer = line
            if (code := codeop.compile_command(line + '\n', '<stdin>', 'single')) is None:
                # Multi-line mode
                with self._enable_multiline_completion():
                            except (EOFError, KeyboardInterrupt):
                                self.lastcmd = ""
                                print('\n')
                                return
                                return None, None, False
                        else:
                            self.stdout.write(continue_prompt)
                            self.stdout.flush()
                                self.lastcmd = ""
                                self.stdout.write('\n')
                                self.stdout.flush()
                                return
                                return None, None, False
                            else:
                                line = line.rstrip('\r\n')
                        buffer += '\n' + line
                    self.lastcmd = buffer
        except SyntaxError as e:
            # Maybe it's an await expression/statement
            if (
                self.async_shim_frame is not None
                and e.msg == "'await' outside function"
            ):
                is_await_code = True
            else:
                raise

        return code, buffer, is_await_code

    def default(self, line):
        if line[:1] == '!': line = line[1:].strip()
        locals = self.curframe.f_locals
        globals = self.curframe.f_globals
        try:
            code, buffer, is_await_code = self._read_code(line)
            if buffer is None:
                return
            save_stdout = sys.stdout
            save_stdin = sys.stdin
            save_displayhook = sys.displayhook
            try:
                sys.stdin = self.stdin
                sys.stdout = self.stdout
                sys.displayhook = self.displayhook
                if not self._exec_in_closure(buffer, globals, locals):
                    exec(code, globals, locals)
                if is_await_code:
                    self._exec_await(buffer, globals, locals)
                    return True
                else:
                    if not self._exec_in_closure(buffer, globals, locals):
                        exec(code, globals, locals)
            finally:
                sys.stdout = save_stdout
                sys.stdin = save_stdin
        pdb.message(header)
    pdb.set_trace(sys._getframe().f_back, commands=commands)

 async def set_trace_async(*, header=None, commands=None):
    """Enter the debugger at the calling stack frame, but in async mode.

    This should be used as await pdb.set_trace_async(). Users can do await
    if they enter the debugger with this function. Otherwise it's the same
    as set_trace().
    """
    if Pdb._last_pdb_instance is not None:
        pdb = Pdb._last_pdb_instance
    else:
        pdb = Pdb(mode='inline', backend='monitoring')
    if header is not None:
        pdb.message(header)
    await pdb.set_trace_async(sys._getframe().f_back, commands=commands)

 # Remote PDB

 class _PdbServer(Pdb):
Original file line number	Diff line number	Diff line change
Expand Up		@@ -188,6 +188,21 @@ slightly different way:
		.. versionadded:: 3.14
		The commands argument.


		.. awaitablefunction:: set_trace_async(*, header=None, commands=None)

		async version of :func:`set_trace`. This function should be used inside an
		async function with :keyword:`await`.

		.. code-block:: python

		async def f():
		await pdb.set_trace_async()

		:keyword:`await` statements are supported if the debugger is invoked by this function.

		.. versionadded:: 3.14

		.. function:: post_mortem(t=None)

		Enter post-mortem debugging of the given exception or
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -1151,6 +1151,11 @@ pdb
		backend by default, which is configurable.
		(Contributed by Tian Gao in :gh:`124533`.)

		* :func:`pdb.set_trace_async` is added to support debugging asyncio
		coroutines. :keyword:`await` statements are supported with this
		function.
		(Contributed by Tian Gao in :gh:`132576`.)


		pickle
		------
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -385,6 +385,9 @@ def __init__(self, completekey='tab', stdin=None, stdout=None, skip=None,
		self.commands_bnum = None # The breakpoint number for which we are
		# defining a list

		self.async_shim_frame = None
		self.async_awaitable = None

		self._chained_exceptions = tuple()
		self._chained_exception_index = 0

Expand All		@@ -400,6 +403,57 @@ def set_trace(self, frame=None, *, commands=None):

		super().set_trace(frame)

		async def set_trace_async(self, frame=None, *, commands=None):
		if self.async_awaitable is not None:
		# We are already in a set_trace_async call, do not mess with it
		return

		if frame is None:
		frame = sys._getframe().f_back

		# We need set_trace to set up the basics, however, this will call
		# set_stepinstr() will we need to compensate for, because we don't
		# want to trigger on calls
		self.set_trace(frame, commands=commands)
		# Changing the stopframe will disable trace dispatch on calls
		self.stopframe = frame
		# We need to stop tracing because we don't have the privilege to avoid
		# triggering tracing functions as normal, as we are not already in
		# tracing functions
		self.stop_trace()

		self.async_shim_frame = sys._getframe()
		self.async_awaitable = None

		while True:
		self.async_awaitable = None
		# Simulate a trace event
		# This should bring up pdb and make pdb believe it's debugging the
		# caller frame
		self.trace_dispatch(frame, "opcode", None)
		if self.async_awaitable is not None:
		try:
		if self.breaks:
		with self.set_enterframe(frame):
		# set_continue requires enterframe to work
		self.set_continue()
		self.start_trace()
		await self.async_awaitable
		except Exception:
		self._error_exc()
		else:
		break

		self.async_shim_frame = None

		# start the trace (the actual command is already set by set_* calls)
		if self.returnframe is None and self.stoplineno == -1 and not self.breaks:
		# This means we did a continue without any breakpoints, we should not
		# start the trace
		return

		self.start_trace()

		def sigint_handler(self, signum, frame):
		if self.allow_kbdint:
		raise KeyboardInterrupt
Expand DownExpand Up		@@ -782,12 +836,25 @@ def _exec_in_closure(self, source, globals, locals):

		return True

		def default(self, line):
		if line[:1] == '!': line = line[1:].strip()
		locals = self.curframe.f_locals
		globals = self.curframe.f_globals
		def _exec_await(self, source, globals, locals):
		""" Run source code that contains await by playing with async shim frame"""
		# Put the source in an async function
		source_async = (
		"async def __pdb_await():\n" +
		textwrap.indent(source, " ") + '\n' +
		" __pdb_locals.update(locals())"
		)
		ns = globals \| locals
		# We use __pdb_locals to do write back
		ns["__pdb_locals"] = locals
		exec(source_async, ns)
		self.async_awaitable = ns["__pdb_await"]()

		def _read_code(self, line):
		buffer = line
		is_await_code = False
		code = None
		try:
		buffer = line
		if (code := codeop.compile_command(line + '\n', '<stdin>', 'single')) is None:
		# Multi-line mode
		with self._enable_multiline_completion():
Expand All		@@ -800,7 +867,7 @@ def default(self, line):
		except (EOFError, KeyboardInterrupt):
		self.lastcmd = ""
		print('\n')
		return
		return None, None, False
		else:
		self.stdout.write(continue_prompt)
		self.stdout.flush()
Expand All		@@ -809,20 +876,44 @@ def default(self, line):
		self.lastcmd = ""
		self.stdout.write('\n')
		self.stdout.flush()
		return
		return None, None, False
		else:
		line = line.rstrip('\r\n')
		buffer += '\n' + line
		self.lastcmd = buffer
		except SyntaxError as e:
		# Maybe it's an await expression/statement
		if (
		self.async_shim_frame is not None
		and e.msg == "'await' outside function"
		):
		is_await_code = True
		else:
		raise

		return code, buffer, is_await_code

		def default(self, line):
		if line[:1] == '!': line = line[1:].strip()
		locals = self.curframe.f_locals
		globals = self.curframe.f_globals
		try:
		code, buffer, is_await_code = self._read_code(line)
		if buffer is None:
		return
		save_stdout = sys.stdout
		save_stdin = sys.stdin
		save_displayhook = sys.displayhook
		try:
		sys.stdin = self.stdin
		sys.stdout = self.stdout
		sys.displayhook = self.displayhook
		if not self._exec_in_closure(buffer, globals, locals):
		exec(code, globals, locals)
		if is_await_code:
		self._exec_await(buffer, globals, locals)
		return True
		else:
		if not self._exec_in_closure(buffer, globals, locals):
		exec(code, globals, locals)
		finally:
		sys.stdout = save_stdout
		sys.stdin = save_stdin
Expand DownExpand Up		@@ -2501,6 +2592,21 @@ def set_trace(*, header=None, commands=None):
		pdb.message(header)
		pdb.set_trace(sys._getframe().f_back, commands=commands)

		async def set_trace_async(*, header=None, commands=None):
		"""Enter the debugger at the calling stack frame, but in async mode.

		This should be used as await pdb.set_trace_async(). Users can do await
		if they enter the debugger with this function. Otherwise it's the same
		as set_trace().
		"""
		if Pdb._last_pdb_instance is not None:
		pdb = Pdb._last_pdb_instance
		else:
		pdb = Pdb(mode='inline', backend='monitoring')
		if header is not None:
		pdb.message(header)
		await pdb.set_trace_async(sys._getframe().f_back, commands=commands)

		# Remote PDB

		class _PdbServer(Pdb):
Expand Down