"""

The Python Debugger Pdb

=======================

To use the debugger in its simplest form:

>>> import pdb

>>> pdb.run('<a statement>')

The debugger's prompt is '(Pdb) '. This will stop in the first

function call in <a statement>.

Alternatively, if a statement terminated with an unhandled exception,

you can use pdb's post-mortem facility to inspect the contents of the

traceback:

>>> <a statement>

<exception traceback>

>>> import pdb

>>> pdb.pm()

The commands recognized by the debugger are listed in the next

section. Most can be abbreviated as indicated; e.g., h(elp) means

that 'help' can be typed as 'h' or 'help' (but not as 'he' or 'hel',

nor as 'H' or 'Help' or 'HELP'). Optional arguments are enclosed in

square brackets. Alternatives in the command syntax are separated

by a vertical bar (|).

A blank line repeats the previous command literally, except for

'list', where it lists the next 11 lines.

Commands that the debugger doesn't recognize are assumed to be Python

statements and are executed in the context of the program being

debugged. Python statements can also be prefixed with an exclamation

point ('!'). This is a powerful way to inspect the program being

debugged; it is even possible to change variables or call functions.

When an exception occurs in such a statement, the exception name is

printed but the debugger's state is not changed.

The debugger supports aliases, which can save typing. And aliases can

have parameters (see the alias help entry) which allows one a certain

level of adaptability to the context under examination.

Multiple commands may be entered on a single line, separated by the

pair ';;'. No intelligence is applied to separating the commands; the

input is split at the first ';;', even if it is in the middle of a

quoted string.

If a file ".pdbrc" exists in your home directory or in the current

directory, it is read in and executed as if it had been typed at the

debugger prompt. This is particularly useful for aliases. If both

files exist, the one in the home directory is read first and aliases

defined there can be overridden by the local file. This behavior can be

disabled by passing the "readrc=False" argument to the Pdb constructor.

Aside from aliases, the debugger is not directly programmable; but it

is implemented as a class from which you can derive your own debugger

class, which you can make as fancy as you like.

Debugger commands

=================

"""

# NOTE: the actual command documentation is collected from docstrings of the

# commands and is appended to __doc__ after the class has been defined.

import os

import io

import re

import sys

import cmd

import bdb

import dis

import code

import glob

import json

import stat

import token

import types

import atexit

import codeop

import pprint

import signal

import socket

import typing

import asyncio

import inspect

import weakref

import builtins

import tempfile

import textwrap

import tokenize

import itertools

import traceback

import linecache

import selectors

import threading

import _colorize

import _pyrepl.utils

from contextlib import ExitStack, closing, contextmanager

from types import CodeType

from warnings import deprecated

class Restart(Exception):

"""Causes a debugger to be restarted for the debugged python program."""

pass

__all__ = ["run", "pm", "Pdb", "runeval", "runctx", "runcall", "set_trace",

"post_mortem", "set_default_backend", "get_default_backend", "help"]

def find_first_executable_line(code):

""" Try to find the first executable line of the code object.

Equivalently, find the line number of the instruction that's

after RESUME

Return code.co_firstlineno if no executable line is found.

"""

prev = None

for instr in dis.get_instructions(code):

if prev is not None and prev.opname == 'RESUME':

if instr.positions.lineno is not None:

return instr.positions.lineno

return code.co_firstlineno

prev = instr

return code.co_firstlineno

def find_function(funcname, filename):

cre = re.compile(r'(?:async\s+)?def\s+%s(\s*\[.+\])?\s*[(]' % re.escape(funcname))

try:

fp = tokenize.open(filename)

except OSError:

lines = linecache.getlines(filename)

if not lines:

return None

fp = io.StringIO(''.join(lines))

funcdef = ""

funcstart = 0

# consumer of this info expects the first line to be 1

with fp:

for lineno, line in enumerate(fp, start=1):

if cre.match(line):

funcstart, funcdef = lineno, line

elif funcdef:

funcdef += line

if funcdef:

try:

code = compile(funcdef, filename, 'exec')

except SyntaxError:

continue

# We should always be able to find the code object here

funccode = next(c for c in code.co_consts if

isinstance(c, CodeType) and c.co_name == funcname)

lineno_offset = find_first_executable_line(funccode)

return funcname, filename, funcstart + lineno_offset - 1

return None

def lasti2lineno(code, lasti):

linestarts = list(dis.findlinestarts(code))

linestarts.reverse()

for i, lineno in linestarts:

if lasti >= i:

return lineno

return 0

class _rstr(str):

"""String that doesn't quote its repr."""

def __repr__(self):

return self

class _ExecutableTarget:

filename: str

code: CodeType | str

namespace: dict

class _ScriptTarget(_ExecutableTarget):

def __init__(self, target):

self._target = os.path.realpath(target)

if not os.path.exists(self._target):

print(f'Error: {target} does not exist')

sys.exit(1)

if os.path.isdir(self._target):

print(f'Error: {target} is a directory')

sys.exit(1)

# If safe_path(-P) is not set, sys.path[0] is the directory

# of pdb, and we should replace it with the directory of the script

if not sys.flags.safe_path:

sys.path[0] = os.path.dirname(self._target)

def __repr__(self):

return self._target

@property

def filename(self):

return self._target

@property

def code(self):

# Open the file each time because the file may be modified

with io.open_code(self._target) as fp:

return f"exec(compile({fp.read()!r}, {self._target!r}, 'exec'))"

@property

def namespace(self):

return dict(

__name__='__main__',

__file__=self._target,

__builtins__=__builtins__,

__spec__=None,

)

class _ModuleTarget(_ExecutableTarget):

def __init__(self, target):

self._target = target

import runpy

try:

_, self._spec, self._code = runpy._get_module_details(self._target)

except ImportError as e:

print(f"ImportError: {e}")

sys.exit(1)

except Exception:

traceback.print_exc()

sys.exit(1)

def __repr__(self):

return self._target

@property

def filename(self):

return self._code.co_filename

@property

def code(self):

return self._code

@property

def namespace(self):

return dict(

__name__='__main__',

__file__=os.path.normcase(os.path.abspath(self.filename)),

__package__=self._spec.parent,

__loader__=self._spec.loader,

__spec__=self._spec,

__builtins__=__builtins__,

)

class _ZipTarget(_ExecutableTarget):

def __init__(self, target):

import runpy

self._target = os.path.realpath(target)

sys.path.insert(0, self._target)

try:

_, self._spec, self._code = runpy._get_main_module_details()

except ImportError as e:

print(f"ImportError: {e}")

sys.exit(1)

except Exception:

traceback.print_exc()

sys.exit(1)

def __repr__(self):

return self._target

@property

def filename(self):

return self._code.co_filename

@property

def code(self):

return self._code

@property

def namespace(self):

return dict(

__name__='__main__',

__file__=os.path.normcase(os.path.abspath(self.filename)),

__package__=self._spec.parent,

__loader__=self._spec.loader,

__spec__=self._spec,

__builtins__=__builtins__,

)

class _PdbInteractiveConsole(code.InteractiveConsole):

def __init__(self, ns, message):

self._message = message

super().__init__(locals=ns, local_exit=True)

def write(self, data):

self._message(data, end='')

# Interaction prompt line will separate file and call info from code

# text using value of line_prefix string. A newline and arrow may

# be to your liking. You can set it once pdb is imported using the

# command "pdb.line_prefix = '\n% '".

# line_prefix = ': ' # Use this to get the old situation back

line_prefix = '\n-> ' # Probably a better default

# The default backend to use for Pdb instances if not specified

# Should be either 'settrace' or 'monitoring'

_default_backend = 'settrace'

def set_default_backend(backend):

"""Set the default backend to use for Pdb instances."""

global _default_backend

if backend not in ('settrace', 'monitoring'):

raise ValueError("Invalid backend: %s" % backend)

_default_backend = backend

def get_default_backend():

"""Get the default backend to use for Pdb instances."""

return _default_backend

class Pdb(bdb.Bdb, cmd.Cmd):

_previous_sigint_handler = None

# Limit the maximum depth of chained exceptions, we should be handling cycles,

# but in case there are recursions, we stop at 999.

MAX_CHAINED_EXCEPTION_DEPTH = 999

_file_mtime_table = {}

_last_pdb_instance = None

def __init__(self, completekey='tab', stdin=None, stdout=None, skip=None,

nosigint=False, readrc=True, mode=None, backend=None, colorize=False):

bdb.Bdb.__init__(self, skip=skip, backend=backend if backend else get_default_backend())

cmd.Cmd.__init__(self, completekey, stdin, stdout)

sys.audit("pdb.Pdb")

if stdin:

self.use_rawinput = False

self.prompt = '(Pdb) '

self.aliases = {}

self.displaying = {}

self.mainpyfile = ''

self._wait_for_mainpyfile = False

self.tb_lineno = {}

self.mode = mode

self.colorize = colorize and _colorize.can_colorize(file=stdout or sys.stdout)

# Try to load readline if it exists

try:

import readline

# remove some common file name delimiters

readline.set_completer_delims(' \t\n`@#%^&*()=+[{]}\\|;:\'",<>?')

except ImportError:

pass

self.allow_kbdint = False

self.nosigint = nosigint

# Consider these characters as part of the command so when the users type

# c.a or c['a'], it won't be recognized as a c(ontinue) command

self.identchars = cmd.Cmd.identchars + '=.[](),"\'+-*/%@&|<>~^'

# Read ~/.pdbrc and ./.pdbrc

self.rcLines = []

if readrc:

try:

with open(os.path.expanduser('~/.pdbrc'), encoding='utf-8') as rcFile:

self.rcLines.extend(rcFile)

except OSError:

pass

try:

with open(".pdbrc", encoding='utf-8') as rcFile:

self.rcLines.extend(rcFile)

except OSError:

pass

self.commands = {} # associates a command list to breakpoint numbers

self.commands_defining = False # True while in the process of defining

# a command list

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

self._current_task = None

self.lineno = None

self.stack = []

self.curindex = 0

self.curframe = None

self._user_requested_quit = False

def set_trace(self, frame=None, *, commands=None):

Pdb._last_pdb_instance = self

if frame is None:

frame = sys._getframe().f_back

if commands is not None:

self.rcLines.extend(commands)

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

self.message("\nProgram interrupted. (Use 'cont' to resume).")

self.set_step()

self.set_trace(frame)

def reset(self):

bdb.Bdb.reset(self)

self.forget()

def forget(self):

self.lineno = None

self.stack = []

self.curindex = 0

if self.curframe:

self.curframe.f_globals.pop('__pdb_convenience_variables', None)

self.curframe = None

self.tb_lineno.clear()

def setup(self, f, tb):

self.forget()

self.stack, self.curindex = self.get_stack(f, tb)

while tb:

# when setting up post-mortem debugging with a traceback, save all

# the original line numbers to be displayed along the current line

# numbers (which can be different, e.g. due to finally clauses)

lineno = lasti2lineno(tb.tb_frame.f_code, tb.tb_lasti)

self.tb_lineno[tb.tb_frame] = lineno

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:

self.set_convenience_variable(

self.curframe,

'_exception',

self._chained_exceptions[self._chained_exception_index],

)

if self.rcLines:

self.cmdqueue = [

line for line in self.rcLines

if line.strip() and not line.strip().startswith("#")

]

self.rcLines = []

@property

@deprecated("The frame locals reference is no longer cached. Use 'curframe.f_locals' instead.")

def curframe_locals(self):

return self.curframe.f_locals

@curframe_locals.setter

@deprecated("Setting 'curframe_locals' no longer has any effect. Update the contents of 'curframe.f_locals' instead.")

def curframe_locals(self, value):

pass

# Override Bdb methods

def user_call(self, frame, argument_list):

"""This method is called when there is the remote possibility

that we ever need to stop in this function."""

if self._wait_for_mainpyfile:

return

if self.stop_here(frame):

self.message('--Call--')

self.interaction(frame, None)

def user_line(self, frame):

"""This function is called when we stop or break at this line."""

if self._wait_for_mainpyfile:

if (self.mainpyfile != self.canonic(frame.f_code.co_filename)):

return

self._wait_for_mainpyfile = False

if self.trace_opcodes:

# GH-127321

# We want to avoid stopping at an opcode that does not have

# an associated line number because pdb does not like it

if frame.f_lineno is None:

self.set_stepinstr()

return

self.bp_commands(frame)

self.interaction(frame, None)

user_opcode = user_line

def bp_commands(self, frame):

"""Call every command that was set for the current active breakpoint

(if there is one).

Returns True if the normal interaction function must be called,

False otherwise."""

# self.currentbp is set in bdb in Bdb.break_here if a breakpoint was hit

if getattr(self, "currentbp", False) and \

self.currentbp in self.commands:

currentbp = self.currentbp

self.currentbp = 0

for line in self.commands[currentbp]:

self.cmdqueue.append(line)

self.cmdqueue.append(f'_pdbcmd_restore_lastcmd {self.lastcmd}')

def user_return(self, frame, return_value):

"""This function is called when a return trap is set here."""

if self._wait_for_mainpyfile:

return

frame.f_locals['__return__'] = return_value

self.set_convenience_variable(frame, '_retval', return_value)

self.message('--Return--')

self.interaction(frame, None)

def user_exception(self, frame, exc_info):

"""This function is called if an exception occurs,

but only if we are to stop at or just below this level."""

if self._wait_for_mainpyfile:

return

exc_type, exc_value, exc_traceback = exc_info

frame.f_locals['__exception__'] = exc_type, exc_value

self.set_convenience_variable(frame, '_exception', exc_value)

# An 'Internal StopIteration' exception is an exception debug event

# issued by the interpreter when handling a subgenerator run with

# 'yield from' or a generator controlled by a for loop. No exception has

# actually occurred in this case. The debugger uses this debug event to

# stop when the debuggee is returning from such generators.

prefix = 'Internal ' if (not exc_traceback

and exc_type is StopIteration) else ''

self.message('%s%s' % (prefix, self._format_exc(exc_value)))

self.interaction(frame, exc_traceback)

# General interaction function

def _cmdloop(self):

while True:

try:

# keyboard interrupts allow for an easy way to cancel

# the current command, so allow them during interactive input

self.allow_kbdint = True

self.cmdloop()

self.allow_kbdint = False

break

except KeyboardInterrupt:

self.message('--KeyboardInterrupt--')

def _save_initial_file_mtime(self, frame):

"""save the mtime of the all the files in the frame stack in the file mtime table

if they haven't been saved yet."""

while frame:

filename = frame.f_code.co_filename

if filename not in self._file_mtime_table:

try:

self._file_mtime_table[filename] = os.path.getmtime(filename)

except Exception:

pass

frame = frame.f_back

def _validate_file_mtime(self):

"""Check if the source file of the current frame has been modified.

If so, give a warning and reset the modify time to current."""

try:

filename = self.curframe.f_code.co_filename

mtime = os.path.getmtime(filename)

except Exception:

return

if (filename in self._file_mtime_table and

mtime != self._file_mtime_table[filename]):

self.message(f"*** WARNING: file '{filename}' was edited, "

"running stale code until the program is rerun")

self._file_mtime_table[filename] = mtime

# Called before loop, handles display expressions

# Set up convenience variable containers

def _show_display(self):

displaying = self.displaying.get(self.curframe)

if displaying:

for expr, oldvalue in displaying.items():

newvalue = self._getval_except(expr)

# check for identity first; this prevents custom __eq__ to

# be called at every loop, and also prevents instances whose

# fields are changed to be displayed

if newvalue is not oldvalue and newvalue != oldvalue:

displaying[expr] = newvalue

self.message('display %s: %s [old: %s]' %

(expr, self._safe_repr(newvalue, expr),

self._safe_repr(oldvalue, expr)))

def _get_tb_and_exceptions(self, tb_or_exc):

"""

Given a traceback or an exception, return a tuple of chained exceptions

and current traceback to inspect.

This will deal with selecting the right ``__cause__`` or ``__context__``

as well as handling cycles, and return a flattened list of exceptions we

can jump to with do_exceptions.

"""

_exceptions = []

if isinstance(tb_or_exc, BaseException):

traceback, current = tb_or_exc.__traceback__, tb_or_exc

while current is not None:

if current in _exceptions:

break

_exceptions.append(current)

if current.__cause__ is not None:

current = current.__cause__

elif (

current.__context__ is not None and not current.__suppress_context__

):

current = current.__context__

if len(_exceptions) >= self.MAX_CHAINED_EXCEPTION_DEPTH:

self.message(

f"More than {self.MAX_CHAINED_EXCEPTION_DEPTH}"

" chained exceptions found, not all exceptions"

"will be browsable with `exceptions`."

)

break

else:

traceback = tb_or_exc

return tuple(reversed(_exceptions)), traceback

@contextmanager

def _hold_exceptions(self, exceptions):

"""

Context manager to ensure proper cleaning of exceptions references

When given a chained exception instead of a traceback,

pdb may hold references to many objects which may leak memory.

We use this context manager to make sure everything is properly cleaned

"""

try:

self._chained_exceptions = exceptions

self._chained_exception_index = len(exceptions) - 1

yield

finally:

# we can't put those in forget as otherwise they would

# be cleared on exception change

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:

try:

signal.signal(signal.SIGINT, Pdb._previous_sigint_handler)

except ValueError: # ValueError: signal only works in main thread

pass

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"

with self._hold_exceptions(_chained_exceptions):

self.setup(frame, tb)

# We should print the stack entry if and only if the user input

# is expected, and we should print it right before the user input.

# We achieve this by appending _pdbcmd_print_frame_status to the

# command queue. If cmdqueue is not exhausted, the user input is

# not expected and we will not print the stack entry.

self.cmdqueue.append('_pdbcmd_print_frame_status')

self._cmdloop()

# If _pdbcmd_print_frame_status is not used, pop it out

if self.cmdqueue and self.cmdqueue[-1] == '_pdbcmd_print_frame_status':

self.cmdqueue.pop()

self.forget()

def displayhook(self, obj):

"""Custom displayhook for the exec in default(), which prevents

assignment of the _ variable in the builtins.

"""

# reproduce the behavior of the standard displayhook, not printing None

if obj is not None:

self.message(repr(obj))

@contextmanager

def _enable_multiline_input(self):

try:

import readline

except ImportError:

yield

return

def input_auto_indent():

last_index = readline.get_current_history_length()

last_line = readline.get_history_item(last_index)

if last_line:

if last_line.isspace():

# If the last line is empty, we don't need to indent

return

last_line = last_line.rstrip('\r\n')

indent = len(last_line) - len(last_line.lstrip())

if last_line.endswith(":"):

indent += 4

readline.insert_text(' ' * indent)

completenames = self.completenames

try:

self.completenames = self.complete_multiline_names

readline.set_startup_hook(input_auto_indent)

yield

finally:

readline.set_startup_hook()

self.completenames = completenames

return

def _exec_in_closure(self, source, globals, locals):

""" Run source code in closure so code object created within source

can find variables in locals correctly

returns True if the source is executed, False otherwise

"""

# Determine if the source should be executed in closure. Only when the

# source compiled to multiple code objects, we should use this feature.

# Otherwise, we can just raise an exception and normal exec will be used.

code = compile(source, "<string>", "exec")

if not any(isinstance(const, CodeType) for const in code.co_consts):

return False

# locals could be a proxy which does not support pop

# copy it first to avoid modifying the original locals

locals_copy = dict(locals)

locals_copy["__pdb_eval__"] = {

"result": None,

"write_back": {}

}

# If the source is an expression, we need to print its value

try:

compile(source, "<string>", "eval")

except SyntaxError:

pass

else:

source = "__pdb_eval__['result'] = " + source

# Add write-back to update the locals

source = ("try:\n" +

textwrap.indent(source, " ") + "\n" +

"finally:\n" +

" __pdb_eval__['write_back'] = locals()")

# Build a closure source code with freevars from locals like:

# def __pdb_outer():

# var = None

# def __pdb_scope(): # This is the code object we want to execute

# nonlocal var

# <source>

# return __pdb_scope.__code__

source_with_closure = ("def __pdb_outer():\n" +

"\n".join(f" {var} = None" for var in locals_copy) + "\n" +

" def __pdb_scope():\n" +

"\n".join(f" nonlocal {var}" for var in locals_copy) + "\n" +

textwrap.indent(source, " ") + "\n" +

" return __pdb_scope.__code__"

)

# Get the code object of __pdb_scope()

# The exec fills locals_copy with the __pdb_outer() function and we can call

# that to get the code object of __pdb_scope()

ns = {}

try:

exec(source_with_closure, {}, ns)

except Exception:

return False

code = ns["__pdb_outer"]()

cells = tuple(types.CellType(locals_copy.get(var)) for var in code.co_freevars)

try:

exec(code, globals, locals_copy, closure=cells)

except Exception:

return False

# get the data we need from the statement

pdb_eval = locals_copy["__pdb_eval__"]

# __pdb_eval__ should not be updated back to locals

pdb_eval["write_back"].pop("__pdb_eval__")

# Write all local variables back to locals

locals.update(pdb_eval["write_back"])

eval_result = pdb_eval["result"]

if eval_result is not None:

print(repr(eval_result))

return True

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:

if (code := codeop.compile_command(line + '\n', '<stdin>', 'single')) is None:

# Multi-line mode

with self._enable_multiline_input():

buffer = line

continue_prompt = "... "

while (code := codeop.compile_command(buffer, '<stdin>', 'single')) is None:

if self.use_rawinput:

try:

line = input(continue_prompt)

except (EOFError, KeyboardInterrupt):

self.lastcmd = ""

print('\n')

return None, None, False

else:

self.stdout.write(continue_prompt)

self.stdout.flush()

line = self.stdin.readline()

if not len(line):

self.lastcmd = ""

self.stdout.write('\n')

self.stdout.flush()

return None, None, False

else:

line = line.rstrip('\r\n')

if line.isspace():

# empty line, just continue

buffer += '\n'

else:

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 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

sys.displayhook = save_displayhook

except:

self._error_exc()

def _replace_convenience_variables(self, line):

"""Replace the convenience variables in 'line' with their values.

e.g. $foo is replaced by __pdb_convenience_variables["foo"].

Note: such pattern in string literals will be skipped"""

if "$" not in line:

return line

dollar_start = dollar_end = (-1, -1)

replace_variables = []

try:

for t in tokenize.generate_tokens(io.StringIO(line).readline):

token_type, token_string, start, end, _ = t

if token_type == token.OP and token_string == '$':

dollar_start, dollar_end = start, end

elif start == dollar_end and token_type == token.NAME:

# line is a one-line command so we only care about column

replace_variables.append((dollar_start[1], end[1], token_string))

except tokenize.TokenError:

return line

if not replace_variables:

return line

last_end = 0

line_pieces = []

for start, end, name in replace_variables:

line_pieces.append(line[last_end:start] + f'__pdb_convenience_variables["{name}"]')

last_end = end

line_pieces.append(line[last_end:])

return ''.join(line_pieces)

def precmd(self, line):

"""Handle alias expansion and ';;' separator."""

if not line.strip():

return line

args = line.split()

while args[0] in self.aliases:

line = self.aliases[args[0]]