def __init__(self, locals=None, encoding=None):

"""The syntaxerror callback can be set at any time and will be called

self.encoding = encoding or sys.getdefaultencoding()

self.syntaxerror_callback = None

# Unfortunately code.InteractiveInterpreter is a classic class, so no super()

code.InteractiveInterpreter.__init__(self, locals)

if not py3:

def runsource(self, source, filename='<input>', symbol='single',

encode=True):

if encode:

source = '# coding: %s\n%s' % (self.encoding,

source.encode(self.encoding))

return code.InteractiveInterpreter.runsource(self, source,

filename, symbol)

def showsyntaxerror(self, filename=None):

"""Override the regular handler, the code's copied and pasted from

if self.syntaxerror_callback is not None:

self.syntaxerror_callback()

type, value, sys.last_traceback = sys.exc_info()

sys.last_type = type

sys.last_value = value

if filename and type is SyntaxError:

# Work hard to stuff the correct filename in the exception

try:

msg, (dummy_filename, lineno, offset, line) = value.args

except:

# Not the format we expect; leave it alone

pass

else:

# Stuff in the right filename and right lineno

if not py3:

lineno -= 1

value = SyntaxError(msg, (filename, lineno, offset, line))

sys.last_value = value

list = traceback.format_exception_only(type, value)

self.writetb(list)

def showtraceback(self):

"""This needs to override the default traceback thing

try:

t, v, tb = sys.exc_info()

sys.last_type = t

sys.last_value = v

sys.last_traceback = tb

tblist = traceback.extract_tb(tb)

del tblist[:1]

# Set the right lineno (encoding header adds an extra line)

if not py3:

for i, (filename, lineno, module, something) in enumerate(tblist):

if filename == '<input>':

tblist[i] = (filename, lineno - 1, module, something)

l = traceback.format_list(tblist)

if l:

l.insert(0, "Traceback (most recent call last):\n")

l[len(l):] = traceback.format_exception_only(t, v)

finally:

tblist = tb = None

self.writetb(l)

def writetb(self, lines):

"""This outputs the traceback and should be overridden for anything

for line in lines:

"""Stores readline-style history and current place in it"""

def __init__(self, entries=None, duplicates=False):

if entries is None:

self.entries = ['']

else:

self.entries = list(entries)

self.index = 0 # how many lines back in history is currently selected

# where 0 is the saved typed line, 1 the prev entered line

self.saved_line = '' # what was on the prompt before using history

self.duplicates = duplicates

def append(self, line):

line = line.rstrip('\n')

if line:

if not self.duplicates:

# remove duplicates

try:

while True:

self.entries.remove(line)

except ValueError:

pass

self.entries.append(line)

def first(self):

"""Move back to the beginning of the history."""

if not self.is_at_end:

self.index = len(self.entries)

return self.entries[-self.index]

def back(self, start=True, search=False, target=None, include_current=False):

"""Move one step back in the history."""

if target is None:

target = self.saved_line

if not self.is_at_end:

if search:

self.index += self.find_partial_match_backward(target, include_current)

elif start:

self.index += self.find_match_backward(target, include_current)

else:

self.index += 1

return self.entry

@property

def entry(self):

"""The current entry, which may be the saved line"""

return self.entries[-self.index] if self.index else self.saved_line

@property

def entries_by_index(self):

return list(reversed(self.entries + [self.saved_line]))

def find_match_backward(self, search_term, include_current=False):

for idx, val in enumerate(self.entries_by_index[self.index + (0 if include_current else 1):]):

if val.startswith(search_term):

return idx + (0 if include_current else 1)

return 0

def find_partial_match_backward(self, search_term, include_current=False):

for idx, val in enumerate(self.entries_by_index[self.index + (0 if include_current else 1):]):

if search_term in val:

return idx + (0 if include_current else 1)

return 0

def forward(self, start=True, search=False, target=None, include_current=False):

"""Move one step forward in the history."""

if target is None:

target = self.saved_line

if self.index > 1:

if search:

self.index -= self.find_partial_match_forward(target, include_current)

elif start:

self.index -= self.find_match_forward(target, include_current)

else:

self.index -= 1

return self.entry

else:

self.index = 0

return self.saved_line

def find_match_forward(self, search_term, include_current=False):

#TODO these are no longer efficient, because we realize the whole list. Does this matter?

for idx, val in enumerate(reversed(self.entries_by_index[:max(0, self.index - (1 if include_current else 0))])):

if val.startswith(search_term):

return idx + (0 if include_current else 1)

return self.index

def find_partial_match_forward(self, search_term, include_current=False):

for idx, val in enumerate(reversed(self.entries_by_index[:max(0, self.index - (1 if include_current else 0))])):

if search_term in val:

return idx + (0 if include_current else 1)

return self.index

def last(self):

"""Move forward to the end of the history."""

if not self.is_at_start:

self.index = 0

return self.entries[0]

@property

def is_at_end(self):

return self.index >= len(self.entries) or self.index == -1

@property

def is_at_start(self):

return self.index == 0

def enter(self, line):

if self.index == 0:

self.saved_line = line

@classmethod

def from_filename(cls, filename):

history = cls()

history.load(filename)

return history

def load(self, filename, encoding):

with codecs.open(filename, 'r', encoding, 'ignore') as hfile:

for line in hfile:

self.append(line)

def reset(self):

self.index = 0

self.saved_line = ''

def save(self, filename, encoding, lines=0):

with codecs.open(filename, 'w', encoding, 'ignore') as hfile:

for line in self.entries[-lines:]:

hfile.write(line)

"""Stores a list of matches and which one is currently selected if any.

def __init__(self):

self.current_word = '' # word being replaced in the original line of text

self.matches = None # possible replacements for current_word

self.index = -1 # which word is currently replacing the current word

self.orig_cursor_offset = None # cursor position in the original line

self.orig_line = None # original line (before match replacements)

self.completer = None # class describing the current type of completion

def __nonzero__(self):

"""MatchesIterator is False when word hasn't been replaced yet"""

return self.index != -1

def __iter__(self):

return self

def current(self):

if self.index == -1:

raise ValueError('No current match.')

return self.matches[self.index]

def next(self):

self.index = (self.index + 1) % len(self.matches)

return self.matches[self.index]

def previous(self):

if self.index <= 0:

self.index = len(self.matches)

self.index -= 1

return self.matches[self.index]

def cur_line(self):

"""Returns a cursor offset and line with the current substitution made"""

return self.substitute(self.current())

def substitute(self, match):

"""Returns a cursor offset and line with match substituted in"""

start, end, word = self.completer.locate(self.orig_cursor_offset, self.orig_line)

result = start + len(match), self.orig_line[:start] + match + self.orig_line[end:]

return result

def is_cseq(self):

return bool(os.path.commonprefix(self.matches)[len(self.current_word):])

def substitute_cseq(self):

"""Returns a new line by substituting a common sequence in, and update matches"""

cseq = os.path.commonprefix(self.matches)

new_cursor_offset, new_line = self.substitute(cseq)

if len(self.matches) == 1:

self.clear()

else:

self.update(new_cursor_offset, new_line, self.matches, self.completer)

if len(self.matches) == 1:

self.clear()

return new_cursor_offset, new_line

def update(self, cursor_offset, current_line, matches, completer):

"""Called to reset the match index and update the word being replaced

self.orig_cursor_offset = cursor_offset

self.orig_line = current_line

assert matches is not None

self.matches = matches

self.completer = completer

#assert self.completer.locate(self.orig_cursor_offset, self.orig_line) is not None, (self.completer.locate, self.orig_cursor_offset, self.orig_line)

self.index = -1

self.start, self.end, self.current_word = self.completer.locate(self.orig_cursor_offset, self.orig_line)

def clear(self):

self.matches = []

self.cursor_offset = -1

self.current_line = ''

self.current_word = ''

self.start = None

self.end = None

def __init__(self, config, statusbar=None):

self.config = config

if statusbar:

self.statusbar = statusbar

def confirm(self, s):

raise NotImplementedError

def notify(self, s, n=10):

raise NotImplementedError

def file_prompt(self, s):

"""Implements the necessary guff for a Python-repl-alike interface

def __init__(self, interp, config):

"""Initialise the repl.

self.config = config

self.cut_buffer = ''

self.buffer = []

self.interp = interp

self.interp.syntaxerror_callback = self.clear_current_line

self.match = False

self.rl_history = History(duplicates=config.hist_duplicates)

self.s_hist = []

self.history = []

self.evaluating = False

self.matches_iter = MatchesIterator()

self.argspec = None

self.current_func = None

self.highlighted_paren = None

self._C = {}

self.prev_block_finished = 0

self.interact = Interaction(self.config)

# previous pastebin content to prevent duplicate pastes, filled on call

# to repl.pastebin

self.prev_pastebin_content = ''

self.prev_pastebin_url = ''

# Necessary to fix mercurial.ui.ui expecting sys.stderr to have this

# attribute

self.closed = False

pythonhist = os.path.expanduser(self.config.hist_file)

if os.path.exists(pythonhist):

self.rl_history.load(pythonhist,

getpreferredencoding() or "ascii")

@property

def ps1(self):

try:

return str(sys.ps1)

except AttributeError:

return '>>> '

@property

def ps2(self):

try:

return str(sys.ps2)

except AttributeError:

return '... '

def startup(self):

"""

filename = os.environ.get('PYTHONSTARTUP')

if filename and os.path.isfile(filename):

with open(filename, 'r') as f:

if py3:

self.interp.runsource(f.read(), filename, 'exec')

else:

self.interp.runsource(f.read(), filename, 'exec', encode=False)

def current_string(self, concatenate=False):

"""If the line ends in a string get it, otherwise return ''"""

tokens = self.tokenize(self.current_line)

string_tokens = list(takewhile(token_is_any_of([Token.String,

Token.Text]),

reversed(tokens)))

if not string_tokens:

return ''

opening = string_tokens.pop()[1]

string = list()

for (token, value) in reversed(string_tokens):

if token is Token.Text:

continue

elif opening is None:

opening = value

elif token is Token.String.Doc:

string.append(value[3:-3])

opening = None

elif value == opening:

opening = None

if not concatenate:

string = list()

else:

string.append(value)

if opening is None:

return ''

return ''.join(string)

def get_object(self, name):

attributes = name.split('.')

obj = eval(attributes.pop(0), self.interp.locals)

while attributes:

with inspection.AttrCleaner(obj):

obj = getattr(obj, attributes.pop(0))

return obj

def get_args(self):

"""Check if an unclosed parenthesis exists, then attempt to get the

self.current_func = None

if not self.config.arg_spec:

return False

# Get the name of the current function and where we are in

# the arguments

stack = [['', 0, '']]

try:

for (token, value) in PythonLexer().get_tokens(

self.current_line):

if token is Token.Punctuation:

if value in '([{':

stack.append(['', 0, value])

elif value in ')]}':

stack.pop()

elif value == ',':

try:

stack[-1][1] += 1

except TypeError:

stack[-1][1] = ''

stack[-1][0] = ''

elif value == ':' and stack[-1][2] == 'lambda':

stack.pop()

else:

stack[-1][0] = ''

elif (token is Token.Name or token in Token.Name.subtypes or

token is Token.Operator and value == '.'):

stack[-1][0] += value

elif token is Token.Operator and value == '=':

stack[-1][1] = stack[-1][0]

stack[-1][0] = ''

elif token is Token.Keyword and value == 'lambda':

stack.append(['', 0, value])

else:

stack[-1][0] = ''

while stack[-1][2] in '[{':

stack.pop()

_, arg_number, _ = stack.pop()

func, _, _ = stack.pop()

except IndexError:

return False

if not func:

return False

try:

f = self.get_object(func)

except Exception:

# another case of needing to catch every kind of error

# since user code is run in the case of descriptors

# XXX: Make sure you raise here if you're debugging the completion

# stuff !

return False

if inspect.isclass(f):

try:

if f.__init__ is not object.__init__:

f = f.__init__

except AttributeError:

return None

self.current_func = f

self.argspec = inspection.getargspec(func, f)

if self.argspec:

self.argspec.append(arg_number)

return True

return False

def get_source_of_current_name(self):

"""Return the source code of the object which is bound to the

try:

obj = self.current_func

if obj is None:

line = self.current_line

if inspection.is_eval_safe_name(line):

obj = self.get_object(line)

if obj is None:

return None

source = inspect.getsource(obj)

except (AttributeError, IOError, NameError, TypeError):

return None

else:

return source

def set_docstring(self):

self.docstring = None

if not self.get_args():

self.argspec = None

elif self.current_func is not None:

try:

self.docstring = pydoc.getdoc(self.current_func)

except IndexError:

self.docstring = None

else:

# pydoc.getdoc() returns an empty string if no

# docstring was found

if not self.docstring:

self.docstring = None

def complete(self, tab=False):

"""Construct a full list of possible completions and construct and

self.set_docstring()

matches, completer = autocomplete.get_completer(

self.cursor_offset,

self.current_line,

self.interp.locals,

self.argspec,

'\n'.join(self.buffer + [self.current_line]),

self.config.autocomplete_mode if hasattr(self.config, 'autocomplete_mode') else autocomplete.SIMPLE,

self.config.complete_magic_methods)

#TODO implement completer.shown_before_tab == False (filenames shouldn't fill screen)

if (matches is None # no completion is relevant

or len(matches) == 0): # a target for completion was found

# but no matches were found

self.matches_iter.clear()

return bool(self.argspec)

self.matches_iter.update(self.cursor_offset,

self.current_line, matches, completer)

if len(matches) == 1:

self.matches_iter.next()

if tab: # if this complete is being run for a tab key press, tab() to do the swap

self.cursor_offset, self.current_line = self.matches_iter.substitute_cseq()

return Repl.complete(self)

elif self.matches_iter.current_word == matches[0]:

self.matches_iter.clear()

return False

return completer.shown_before_tab

else:

assert len(matches) > 1

return tab or completer.shown_before_tab

def format_docstring(self, docstring, width, height):

"""Take a string and try to format it into a sane list of strings to be

lines = docstring.split('\n')

out = []

i = 0

for line in lines:

i += 1

if not line.strip():

out.append('\n')

for block in textwrap.wrap(line, width):

out.append(' ' + block + '\n')

if i >= height:

return out

i += 1

# Drop the last newline

out[-1] = out[-1].rstrip()

return out

def next_indentation(self):

"""Return the indentation of the next line based on the current

if self.buffer:

indentation = next_indentation(self.buffer[-1],

self.config.tab_length)

if indentation and self.config.dedent_after > 0:

line_is_empty = lambda line: not line.strip()

empty_lines = takewhile(line_is_empty, reversed(self.buffer))

if sum(1 for _ in empty_lines) >= self.config.dedent_after:

indentation -= 1

else:

indentation = 0

return indentation

def formatforfile(self, s):

"""Format the stdout buffer to something suitable for writing to disk,

def process():

for line in s.split('\n'):

if line.startswith(self.ps1):

yield line[len(self.ps1):]

elif line.startswith(self.ps2):

yield line[len(self.ps2):]

elif line.rstrip():

yield "# OUT: %s" % (line,)

return "\n".join(process())

def write2file(self):

"""Prompt for a filename and write the current contents of the stdout

try:

fn = self.interact.file_prompt('Save to file (Esc to cancel): ')

if not fn:

self.interact.notify("Save cancelled.")

return

except ValueError:

self.interact.notify("Save cancelled.")

return

if fn.startswith('~'):

fn = os.path.expanduser(fn)

if not fn.endswith('.py') and self.config.save_append_py:

fn = fn + '.py'

mode = 'w'

if os.path.exists(fn):

mode = self.interact.file_prompt('%s already exists. Do you want '

'to (c)ancel, (o)verwrite or '

'(a)ppend? ' % (fn, ))

if mode in ('o', 'overwrite'):

mode = 'w'

elif mode in ('a', 'append'):

mode = 'a'

else:

self.interact.notify('Save cancelled.')

return

s = self.formatforfile(self.getstdout())

try:

f = open(fn, mode)

f.write(s)

f.close()

except IOError:

self.interact.notify("Disk write error for file '%s'." % (fn, ))

else:

self.interact.notify('Saved to %s.' % (fn, ))

def pastebin(self, s=None):

"""Upload to a pastebin and display the URL in the status bar."""

if s is None:

s = self.getstdout()

if (self.config.pastebin_confirm and

not self.interact.confirm(_("Pastebin buffer? (y/N) "))):

self.interact.notify(_("Pastebin aborted"))

return

return self.do_pastebin(s)

def do_pastebin(self, s):

"""Actually perform the upload."""

if s == self.prev_pastebin_content:

self.interact.notify(_('Duplicate pastebin. Previous URL: %s') %

(self.prev_pastebin_url, ))

return self.prev_pastebin_url

if self.config.pastebin_helper:

return self.do_pastebin_helper(s)

else:

return self.do_pastebin_xmlrpc(s)

def do_pastebin_xmlrpc(self, s):

"""Upload to pastebin via XML-RPC."""

try:

pasteservice = ServerProxy(self.config.pastebin_url)

except IOError, e:

self.interact.notify(_("Pastebin error for URL '%s': %s") %

(self.config.pastebin_url, str(e)))

return

self.interact.notify(_('Posting data to pastebin...'))

try:

paste_id = pasteservice.pastes.newPaste('pycon', s, '', '', '',

self.config.pastebin_private)

except (SocketError, XMLRPCError), e:

self.interact.notify(_('Upload failed: %s') % (str(e), ) )

return

self.prev_pastebin_content = s

paste_url_template = Template(self.config.pastebin_show_url)

paste_id = urlquote(paste_id)

paste_url = paste_url_template.safe_substitute(paste_id=paste_id)

self.prev_pastebin_url = paste_url

self.interact.notify(_('Pastebin URL: %s') % (paste_url, ), 10)

return paste_url

def do_pastebin_helper(self, s):

"""Call out to helper program for pastebin upload."""

self.interact.notify(_('Posting data to pastebin...'))

try:

helper = subprocess.Popen('',

executable=self.config.pastebin_helper,

stdin=subprocess.PIPE,

stdout=subprocess.PIPE)

helper.stdin.write(s.encode(getpreferredencoding()))

output = helper.communicate()[0].decode(getpreferredencoding())

paste_url = output.split()[0]

except OSError, e:

if e.errno == errno.ENOENT:

self.interact.notify(_('Upload failed: '

'Helper program not found.'))

else:

self.interact.notify(_('Upload failed: '

'Helper program could not be run.'))

return

if helper.returncode != 0:

self.interact.notify(_('Upload failed: '

'Helper program returned non-zero exit '

'status %s.' % (helper.returncode, )))

return

if not paste_url:

self.interact.notify(_('Upload failed: '

'No output from helper program.'))

return

else:

parsed_url = urlparse(paste_url)

if (not parsed_url.scheme

or any(unicodedata.category(c) == 'Cc' for c in paste_url)):

self.interact.notify(_("Upload failed: "

"Failed to recognize the helper "

"program's output as an URL."))

return

self.prev_pastebin_content = s

self.interact.notify(_('Pastebin URL: %s') % (paste_url, ), 10)

return paste_url

def push(self, s, insert_into_history=True):

"""Push a line of code onto the buffer so it can process it all

s = s.rstrip('\n')

self.buffer.append(s)

if insert_into_history:

self.insert_into_history(s)

more = self.interp.runsource('\n'.join(self.buffer))

if not more:

self.buffer = []

return more

def insert_into_history(self, s):

if self.config.hist_length:

histfilename = os.path.expanduser(self.config.hist_file)

oldhistory = self.rl_history.entries

self.rl_history.entries = []

if os.path.exists(histfilename):

self.rl_history.load(histfilename, getpreferredencoding())

self.rl_history.append(s)

try:

self.rl_history.save(histfilename, getpreferredencoding(), self.config.hist_length)

except EnvironmentError, err:

self.interact.notify("Error occured while writing to file %s (%s) " % (histfilename, err.strerror))

self.rl_history.entries = oldhistory

self.rl_history.append(s)

else:

self.rl_history.append(s)

def undo(self, n=1):

"""Go back in the undo history n steps and call reeavluate()

if not self.history:

return None

if len(self.history) < n:

n = len(self.history)

entries = list(self.rl_history.entries)

self.history = self.history[:-n]

self.reevaluate()

self.rl_history.entries = entries

def flush(self):

"""Olivier Grisel brought it to my attention that the logging

def close(self):

"""See the flush() method docstring."""

def tokenize(self, s, newline=False):

"""Tokenizes a line of code, returning pygments tokens

source = '\n'.join(self.buffer + [s])

cursor = len(source) - self.cpos

if self.cpos:

cursor += 1

stack = list()

all_tokens = list(PythonLexer().get_tokens(source))

# Unfortunately, Pygments adds a trailing newline and strings with

# no size, so strip them

while not all_tokens[-1][1]:

all_tokens.pop()

all_tokens[-1] = (all_tokens[-1][0], all_tokens[-1][1].rstrip('\n'))

line = pos = 0

parens = dict(zip('{([', '})]'))

line_tokens = list()

saved_tokens = list()

search_for_paren = True

for (token, value) in split_lines(all_tokens):

pos += len(value)

if token is Token.Text and value == '\n':

line += 1

# Remove trailing newline

line_tokens = list()

saved_tokens = list()

continue

line_tokens.append((token, value))

saved_tokens.append((token, value))

if not search_for_paren:

continue

under_cursor = (pos == cursor)

if token is Token.Punctuation:

if value in parens:

if under_cursor:

line_tokens[-1] = (Parenthesis.UnderCursor, value)

# Push marker on the stack

stack.append((Parenthesis, value))

else:

stack.append((line, len(line_tokens) - 1,

line_tokens, value))

elif value in parens.itervalues():

saved_stack = list(stack)

try:

while True:

opening = stack.pop()

if parens[opening[-1]] == value:

break

except IndexError:

# SyntaxError.. more closed parentheses than

# opened or a wrong closing paren

opening = None

if not saved_stack:

search_for_paren = False

else:

stack = saved_stack

if opening and opening[0] is Parenthesis:

# Marker found

line_tokens[-1] = (Parenthesis, value)

search_for_paren = False

elif opening and under_cursor and not newline:

if self.cpos:

line_tokens[-1] = (Parenthesis.UnderCursor, value)

else:

# The cursor is at the end of line and next to

# the paren, so it doesn't reverse the paren.