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

hist_size=config.hist_length)

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

self.clipboard = get_clipboard()

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

if os.path.exists(pythonhist):

try:

self.rl_history.load(pythonhist,

getpreferredencoding() or "ascii")

except EnvironmentError:

pass

@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

obj = self.current_func

try:

if obj is None:

line = self.current_line

if not line.strip():

raise SourceNotFound("Nothing to get source of")

if inspection.is_eval_safe_name(line):

obj = self.get_object(line)

return inspect.getsource(obj)

except (AttributeError, NameError) as e:

msg = "Cannot get source: " + str(e)

except IOError as e:

msg = str(e)

except TypeError as e:

if "built-in" in str(e):

msg = "Cannot access source of %r" % (obj, )

else:

msg = "No source code found for %s" % (self.current_line, )

raise SourceNotFound(msg)

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,

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:

with open(fn, mode) as f:

f.write(s)

except IOError:

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

else:

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

def copy2clipboard(self):

"""Copy current content to clipboard."""

if self.clipboard is None:

self.interact.notify(_('No clipboard available.'))

return

content = self.formatforfile(self.getstdout())

try:

self.clipboard.copy(content)

except CopyFailed:

self.interact.notify(_('Could not copy to clipboard.'))

else:

self.interact.notify(_('Copied content to clipboard.'))

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_json(s)

def do_pastebin_json(self, s):

"""Upload to pastebin via json interface."""

url = urljoin(self.config.pastebin_url, '/json/new')

payload = {

'code': s,

'lexer': 'pycon',

'expiry': self.config.pastebin_expiry

}

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

try:

response = requests.post(url, data=payload, verify=True)

response.raise_for_status()

except requests.exceptions.RequestException as exc:

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

return

self.prev_pastebin_content = s

data = response.json()

paste_url_template = Template(self.config.pastebin_show_url)

paste_id = urlquote(data['paste_id'])

paste_url = paste_url_template.safe_substitute(paste_id=paste_id)

removal_url_template = Template(self.config.pastebin_removal_url)

removal_id = urlquote(data['removal_id'])

removal_url = removal_url_template.safe_substitute(removal_id=removal_id)

self.prev_pastebin_url = paste_url

self.interact.notify(_('Pastebin URL: %s - Removal URL: %s') %

(paste_url, removal_url))

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 as 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):

try:

self.rl_history.append_reload_and_write(s, self.config.hist_file,

getpreferredencoding())

except RuntimeError as e:

self.interact.notify(str(e))

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]

if (n == 1 and self.buffer and

hasattr(self, 'take_back_buffer_line')):

self.take_back_buffer_line()

else:

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.

# Therefore, we insert the Parenthesis token

# here instead of the Parenthesis.UnderCursor

# token.

line_tokens[-1] = (Parenthesis, value)

(lineno, i, tokens, opening) = opening

if lineno == len(self.buffer):

self.highlighted_paren = (lineno, saved_tokens)

line_tokens[i] = (Parenthesis, opening)

else:

self.highlighted_paren = (lineno, list(tokens))

# We need to redraw a line

tokens[i] = (Parenthesis, opening)

self.reprint_line(lineno, tokens)

search_for_paren = False

elif under_cursor:

search_for_paren = False

if line != len(self.buffer):

return list()

return line_tokens

def clear_current_line(self):

"""This is used as the exception callback for the Interpreter instance.

def send_to_external_editor(self, text, filename=None):

"""Returns modified text from an editor, or the oriignal text if editor exited with non-zero"""

editor_args = shlex.split(self.config.editor)

with tempfile.NamedTemporaryFile(suffix='.py') as temp:

temp.write(text.encode(getpreferredencoding()))

temp.flush()

if subprocess.call(editor_args + [temp.name]) == 0:

with open(temp.name) as f:

if py3:

return f.read()

else:

return f.read().decode(getpreferredencoding())

else:

return text

def open_in_external_editor(self, filename):

editor_args = shlex.split(self.config.editor)

if subprocess.call(editor_args + [filename]) == 0:

return True

return False

def edit_config(self):

if not (os.path.isfile(self.config.config_path)):

if self.interact.confirm(_("Config file does not exist - create new from default? (y/N)")):

try:

bpython_dir, script_name = os.path.split(__file__)

with open(os.path.join(bpython_dir, "sample-config")) as f:

default_config = f.read()

containing_dir = os.path.dirname(os.path.abspath(self.config.config_path))

if not os.path.exists(containing_dir):

os.makedirs(containing_dir)

with open(self.config.config_path, 'w') as f:

f.write(default_config)

except (IOError, OSError) as e:

self.interact.notify('error creating file: %r' % e)

return False

else:

return False

if self.open_in_external_editor(self.config.config_path):

self.interact.notify('bpython config file edited. Restart bpython for changes to take effect.')

else:

"""Given a code line, return the indentation of the next line."""

line = line.expandtabs(tab_length)

indentation = (len(line) - len(line.lstrip(' '))) // tab_length

if line.rstrip().endswith(':'):

indentation += 1

elif indentation >= 1:

if line.lstrip().startswith(('return', 'pass', 'raise', 'yield')):

indentation -= 1

"""Given a code string s and an initial state inside_string, return

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

if token is Token.String:

value = value.lstrip('bBrRuU')

if value in ['"""', "'''", '"', "'"]:

if not inside_string:

inside_string = value

elif value == inside_string:

inside_string = False

for (token, value) in tokens:

if not value:

continue

while value:

head, newline, value = value.partition('\n')

yield (token, head)

if newline: