Move cast_unicode_py2 calls into HistoryAccessor

IPython/core/history.py

@@ -1,911 +1,912
 """ History related magics and functionality """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import atexit
 import datetime
 import os
 import re
 try:
     import sqlite3
 except ImportError:
     try:
         from pysqlite2 import dbapi2 as sqlite3
     except ImportError:
         sqlite3 = None
 import threading
 
 from traitlets.config.configurable import LoggingConfigurable
 from decorator import decorator
 from IPython.utils.decorators import undoc
 from IPython.utils.path import locate_profile
 from IPython.utils import py3compat
 from traitlets import (
     Any, Bool, Dict, Instance, Integer, List, Unicode, TraitError,
     default, observe,
 )
 from warnings import warn
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 @undoc
 class DummyDB(object):
     """Dummy DB that will act as a black hole for history.
     
     Only used in the absence of sqlite"""
     def execute(*args, **kwargs):
         return []
     
     def commit(self, *args, **kwargs):
         pass
     
     def __enter__(self, *args, **kwargs):
         pass
     
     def __exit__(self, *args, **kwargs):
         pass
 
 
 @decorator
 def needs_sqlite(f, self, *a, **kw):
     """Decorator: return an empty list in the absence of sqlite."""
     if sqlite3 is None or not self.enabled:
         return []
     else:
         return f(self, *a, **kw)
 
 
 if sqlite3 is not None:
     DatabaseError = sqlite3.DatabaseError
     OperationalError = sqlite3.OperationalError
 else:
     @undoc
     class DatabaseError(Exception):
         "Dummy exception when sqlite could not be imported. Should never occur."
     
     @undoc
     class OperationalError(Exception):
         "Dummy exception when sqlite could not be imported. Should never occur."
 
 # use 16kB as threshold for whether a corrupt history db should be saved
 # that should be at least 100 entries or so
 _SAVE_DB_SIZE = 16384
 
 @decorator
 def catch_corrupt_db(f, self, *a, **kw):
     """A decorator which wraps HistoryAccessor method calls to catch errors from
     a corrupt SQLite database, move the old database out of the way, and create
     a new one.
 
     We avoid clobbering larger databases because this may be triggered due to filesystem issues,
     not just a corrupt file.
     """
     try:
         return f(self, *a, **kw)
     except (DatabaseError, OperationalError) as e:
         self._corrupt_db_counter += 1
         self.log.error("Failed to open SQLite history %s (%s).", self.hist_file, e)
         if self.hist_file != ':memory:':
             if self._corrupt_db_counter > self._corrupt_db_limit:
                 self.hist_file = ':memory:'
                 self.log.error("Failed to load history too many times, history will not be saved.")
             elif os.path.isfile(self.hist_file):
                 # move the file out of the way
                 base, ext = os.path.splitext(self.hist_file)
                 size = os.stat(self.hist_file).st_size
                 if size >= _SAVE_DB_SIZE:
                     # if there's significant content, avoid clobbering
                     now = datetime.datetime.now().isoformat().replace(':', '.')
                     newpath = base + '-corrupt-' + now + ext
                     # don't clobber previous corrupt backups
                     for i in range(100):
                         if not os.path.isfile(newpath):
                             break
                         else:
                             newpath = base + '-corrupt-' + now + (u'-%i' % i) + ext
                 else:
                     # not much content, possibly empty; don't worry about clobbering
                     # maybe we should just delete it?
                     newpath = base + '-corrupt' + ext
                 os.rename(self.hist_file, newpath)
                 self.log.error("History file was moved to %s and a new file created.", newpath)
             self.init_db()
             return []
         else:
             # Failed with :memory:, something serious is wrong
             raise
         
 class HistoryAccessorBase(LoggingConfigurable):
     """An abstract class for History Accessors """
 
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         raise NotImplementedError
 
     def search(self, pattern="*", raw=True, search_raw=True,
                output=False, n=None, unique=False):
         raise NotImplementedError
 
     def get_range(self, session, start=1, stop=None, raw=True,output=False):
         raise NotImplementedError
 
     def get_range_by_str(self, rangestr, raw=True, output=False):
         raise NotImplementedError
 
 
 class HistoryAccessor(HistoryAccessorBase):
     """Access the history database without adding to it.
     
     This is intended for use by standalone history tools. IPython shells use
     HistoryManager, below, which is a subclass of this."""
 
     # counter for init_db retries, so we don't keep trying over and over
     _corrupt_db_counter = 0
      # after two failures, fallback on :memory:
     _corrupt_db_limit = 2
 
     # String holding the path to the history file
     hist_file = Unicode(
         help="""Path to file to use for SQLite history database.
         
         By default, IPython will put the history database in the IPython
         profile directory.  If you would rather share one history among
         profiles, you can set this value in each, so that they are consistent.
         
         Due to an issue with fcntl, SQLite is known to misbehave on some NFS
         mounts.  If you see IPython hanging, try setting this to something on a
         local disk, e.g::
         
             ipython --HistoryManager.hist_file=/tmp/ipython_hist.sqlite
 
         you can also use the specific value `:memory:` (including the colon
         at both end but not the back ticks), to avoid creating an history file.
         
         """).tag(config=True)
     
     enabled = Bool(True,
         help="""enable the SQLite history
         
         set enabled=False to disable the SQLite history,
         in which case there will be no stored history, no SQLite connection,
         and no background saving thread.  This may be necessary in some
         threaded environments where IPython is embedded.
         """
     ).tag(config=True)
     
     connection_options = Dict(
         help="""Options for configuring the SQLite connection
         
         These options are passed as keyword args to sqlite3.connect
         when establishing database conenctions.
         """
     ).tag(config=True)
 
     # The SQLite database
     db = Any()
     @observe('db')
     def _db_changed(self, change):
         """validate the db, since it can be an Instance of two different types"""
         new = change['new']
         connection_types = (DummyDB,)
         if sqlite3 is not None:
             connection_types = (DummyDB, sqlite3.Connection)
         if not isinstance(new, connection_types):
             msg = "%s.db must be sqlite3 Connection or DummyDB, not %r" % \
                     (self.__class__.__name__, new)
             raise TraitError(msg)
     
     def __init__(self, profile='default', hist_file=u'', **traits):
         """Create a new history accessor.
         
         Parameters
         ----------
         profile : str
           The name of the profile from which to open history.
         hist_file : str
           Path to an SQLite history database stored by IPython. If specified,
           hist_file overrides profile.
         config : :class:`~traitlets.config.loader.Config`
           Config object. hist_file can also be set through this.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryAccessor, self).__init__(**traits)
         # defer setting hist_file from kwarg until after init,
         # otherwise the default kwarg value would clobber any value
         # set by config
         if hist_file:
             self.hist_file = hist_file
         
         if self.hist_file == u'':
             # No one has set the hist_file, yet.
             self.hist_file = self._get_hist_file_name(profile)
 
         if sqlite3 is None and self.enabled:
             warn("IPython History requires SQLite, your history will not be saved")
             self.enabled = False
         
         self.init_db()
     
     def _get_hist_file_name(self, profile='default'):
         """Find the history file for the given profile name.
         
         This is overridden by the HistoryManager subclass, to use the shell's
         active profile.
         
         Parameters
         ----------
         profile : str
           The name of a profile which has a history file.
         """
         return os.path.join(locate_profile(profile), 'history.sqlite')
     
     @catch_corrupt_db
     def init_db(self):
         """Connect to the database, and create tables if necessary."""
         if not self.enabled:
             self.db = DummyDB()
             return
         
         # use detect_types so that timestamps return datetime objects
         kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
         kwargs.update(self.connection_options)
         self.db = sqlite3.connect(self.hist_file, **kwargs)
         self.db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
                         primary key autoincrement, start timestamp,
                         end timestamp, num_cmds integer, remark text)""")
         self.db.execute("""CREATE TABLE IF NOT EXISTS history
                 (session integer, line integer, source text, source_raw text,
                 PRIMARY KEY (session, line))""")
         # Output history is optional, but ensure the table's there so it can be
         # enabled later.
         self.db.execute("""CREATE TABLE IF NOT EXISTS output_history
                         (session integer, line integer, output text,
                         PRIMARY KEY (session, line))""")
         self.db.commit()
         # success! reset corrupt db count
         self._corrupt_db_counter = 0
 
     def writeout_cache(self):
         """Overridden by HistoryManager to dump the cache before certain
         database lookups."""
         pass
 
     ## -------------------------------
     ## Methods for retrieving history:
     ## -------------------------------
     def _run_sql(self, sql, params, raw=True, output=False):
         """Prepares and runs an SQL query for the history database.
 
         Parameters
         ----------
         sql : str
           Any filtering expressions to go after SELECT ... FROM ...
         params : tuple
           Parameters passed to the SQL query (to replace "?")
         raw, output : bool
           See :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         toget = 'source_raw' if raw else 'source'
         sqlfrom = "history"
         if output:
             sqlfrom = "history LEFT JOIN output_history USING (session, line)"
             toget = "history.%s, output_history.output" % toget
         cur = self.db.execute("SELECT session, line, %s FROM %s " %\
                                 (toget, sqlfrom) + sql, params)
         if output:    # Regroup into 3-tuples, and parse JSON
-            return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
+            return ((ses, lin, (py3compat.cast_unicode_py2(inp), py3compat.cast_unicode_py2(out)))
+                    for ses, lin, inp, out in cur)
         return cur
 
     @needs_sqlite
     @catch_corrupt_db
     def get_session_info(self, session):
         """Get info about a session.
 
         Parameters
         ----------
 
         session : int
             Session number to retrieve.
 
         Returns
         -------
         
         session_id : int
            Session ID number
         start : datetime
            Timestamp for the start of the session.
         end : datetime
            Timestamp for the end of the session, or None if IPython crashed.
         num_cmds : int
            Number of commands run, or None if IPython crashed.
         remark : unicode
            A manually set description.
         """
         query = "SELECT * from sessions where session == ?"
         return self.db.execute(query, (session,)).fetchone()
 
     @catch_corrupt_db
     def get_last_session_id(self):
         """Get the last session ID currently in the database.
         
         Within IPython, this should be the same as the value stored in
         :attr:`HistoryManager.session_number`.
         """
         for record in self.get_tail(n=1, include_latest=True):
             return record[0]
 
     @catch_corrupt_db
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         """Get the last n lines from the history database.
 
         Parameters
         ----------
         n : int
           The number of lines to get
         raw, output : bool
           See :meth:`get_range`
         include_latest : bool
           If False (default), n+1 lines are fetched, and the latest one
           is discarded. This is intended to be used where the function
           is called by a user command, which it should not return.
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         self.writeout_cache()
         if not include_latest:
             n += 1
         cur = self._run_sql("ORDER BY session DESC, line DESC LIMIT ?",
                                 (n,), raw=raw, output=output)
         if not include_latest:
             return reversed(list(cur)[1:])
         return reversed(list(cur))
 
     @catch_corrupt_db
     def search(self, pattern="*", raw=True, search_raw=True,
                output=False, n=None, unique=False):
         """Search the database using unix glob-style matching (wildcards
         * and ?).
 
         Parameters
         ----------
         pattern : str
           The wildcarded pattern to match when searching
         search_raw : bool
           If True, search the raw input, otherwise, the parsed input
         raw, output : bool
           See :meth:`get_range`
         n : None or int
           If an integer is given, it defines the limit of
           returned entries.
         unique : bool
           When it is true, return only unique entries.
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         tosearch = "source_raw" if search_raw else "source"
         if output:
             tosearch = "history." + tosearch
         self.writeout_cache()
         sqlform = "WHERE %s GLOB ?" % tosearch
         params = (pattern,)
         if unique:
             sqlform += ' GROUP BY {0}'.format(tosearch)
         if n is not None:
             sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
             params += (n,)
         elif unique:
             sqlform += " ORDER BY session, line"
         cur = self._run_sql(sqlform, params, raw=raw, output=output)
         if n is not None:
             return reversed(list(cur))
         return cur
     
     @catch_corrupt_db
     def get_range(self, session, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
 
         Parameters
         ----------
         session : int
             Session number to retrieve.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
 
         Returns
         -------
         entries
           An iterator over the desired lines. Each line is a 3-tuple, either
           (session, line, input) if output is False, or
           (session, line, (input, output)) if output is True.
         """
         if stop:
             lineclause = "line >= ? AND line < ?"
             params = (session, start, stop)
         else:
             lineclause = "line>=?"
             params = (session, start)
 
         return self._run_sql("WHERE session==? AND %s" % lineclause,
                                     params, raw=raw, output=output)
 
     def get_range_by_str(self, rangestr, raw=True, output=False):
         """Get lines of history from a string of ranges, as used by magic
         commands %hist, %save, %macro, etc.
 
         Parameters
         ----------
         rangestr : str
           A string specifying ranges, e.g. "5 ~2/1-4". See
           :func:`magic_history` for full details.
         raw, output : bool
           As :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         for sess, s, e in extract_hist_ranges(rangestr):
             for line in self.get_range(sess, s, e, raw=raw, output=output):
                 yield line
 
 
 class HistoryManager(HistoryAccessor):
     """A class to organize all history-related functionality in one place.
     """
     # Public interface
 
     # An instance of the IPython shell we are attached to
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC',
                      allow_none=True)
     # Lists to hold processed and raw history. These start with a blank entry
     # so that we can index them starting from 1
     input_hist_parsed = List([""])
     input_hist_raw = List([""])
     # A list of directories visited during session
     dir_hist = List()
     @default('dir_hist')
     def _dir_hist_default(self):
         try:
             return [py3compat.getcwd()]
         except OSError:
             return []
 
     # A dict of output history, keyed with ints from the shell's
     # execution count.
     output_hist = Dict()
     # The text/plain repr of outputs.
     output_hist_reprs = Dict()
 
     # The number of the current session in the history database
     session_number = Integer()
     
     db_log_output = Bool(False,
         help="Should the history database include output? (default: no)"
     ).tag(config=True)
     db_cache_size = Integer(0,
         help="Write to database every x commands (higher values save disk access & power).\n"
         "Values of 1 or less effectively disable caching."
     ).tag(config=True)
     # The input and output caches
     db_input_cache = List()
     db_output_cache = List()
     
     # History saving in separate thread
     save_thread = Instance('IPython.core.history.HistorySavingThread',
                            allow_none=True)
     try:               # Event is a function returning an instance of _Event...
         save_flag = Instance(threading._Event, allow_none=True)
     except AttributeError:         # ...until Python 3.3, when it's a class.
         save_flag = Instance(threading.Event, allow_none=True)
     
     # Private interface
     # Variables used to store the three last inputs from the user.  On each new
     # history update, we populate the user's namespace with these, shifted as
     # necessary.
     _i00 = Unicode(u'')
     _i = Unicode(u'')
     _ii = Unicode(u'')
     _iii = Unicode(u'')
 
     # A regex matching all forms of the exit command, so that we don't store
     # them in the history (it's annoying to rewind the first entry and land on
     # an exit call).
     _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
 
     def __init__(self, shell=None, config=None, **traits):
         """Create a new history manager associated with a shell instance.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryManager, self).__init__(shell=shell, config=config,
             **traits)
         self.save_flag = threading.Event()
         self.db_input_cache_lock = threading.Lock()
         self.db_output_cache_lock = threading.Lock()
         
         try:
             self.new_session()
         except OperationalError:
             self.log.error("Failed to create history session in %s. History will not be saved.",
                 self.hist_file, exc_info=True)
             self.hist_file = ':memory:'
         
         if self.enabled and self.hist_file != ':memory:':
             self.save_thread = HistorySavingThread(self)
             self.save_thread.start()
 
     def _get_hist_file_name(self, profile=None):
         """Get default history file name based on the Shell's profile.
         
         The profile parameter is ignored, but must exist for compatibility with
         the parent class."""
         profile_dir = self.shell.profile_dir.location
         return os.path.join(profile_dir, 'history.sqlite')
     
     @needs_sqlite
     def new_session(self, conn=None):
         """Get a new session number."""
         if conn is None:
             conn = self.db
         
         with conn:
             cur = conn.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
                             NULL, "") """, (datetime.datetime.now(),))
             self.session_number = cur.lastrowid
             
     def end_session(self):
         """Close the database session, filling in the end time and line count."""
         self.writeout_cache()
         with self.db:
             self.db.execute("""UPDATE sessions SET end=?, num_cmds=? WHERE
                             session==?""", (datetime.datetime.now(),
                             len(self.input_hist_parsed)-1, self.session_number))
         self.session_number = 0
                             
     def name_session(self, name):
         """Give the current session a name in the history database."""
         with self.db:
             self.db.execute("UPDATE sessions SET remark=? WHERE session==?",
                             (name, self.session_number))
                             
     def reset(self, new_session=True):
         """Clear the session history, releasing all object references, and
         optionally open a new session."""
         self.output_hist.clear()
         # The directory history can't be completely empty
         self.dir_hist[:] = [py3compat.getcwd()]
         
         if new_session:
             if self.session_number:
                 self.end_session()
             self.input_hist_parsed[:] = [""]
             self.input_hist_raw[:] = [""]
             self.new_session()
 
     # ------------------------------
     # Methods for retrieving history
     # ------------------------------
     def get_session_info(self, session=0):
         """Get info about a session.
 
         Parameters
         ----------
 
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is the previous session.
 
         Returns
         -------
         
         session_id : int
            Session ID number
         start : datetime
            Timestamp for the start of the session.
         end : datetime
            Timestamp for the end of the session, or None if IPython crashed.
         num_cmds : int
            Number of commands run, or None if IPython crashed.
         remark : unicode
            A manually set description.
         """
         if session <= 0:
             session += self.session_number
 
         return super(HistoryManager, self).get_session_info(session=session)
 
     def _get_range_session(self, start=1, stop=None, raw=True, output=False):
         """Get input and output history from the current session. Called by
         get_range, and takes similar parameters."""
         input_hist = self.input_hist_raw if raw else self.input_hist_parsed
             
         n = len(input_hist)
         if start < 0:
             start += n
         if not stop or (stop > n):
             stop = n
         elif stop < 0:
             stop += n
         
         for i in range(start, stop):
             if output:
                 line = (input_hist[i], self.output_hist_reprs.get(i))
             else:
                 line = input_hist[i]
             yield (0, i, line)
     
     def get_range(self, session=0, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
         
         Parameters
         ----------
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is previous session.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
             
         Returns
         -------
         entries
           An iterator over the desired lines. Each line is a 3-tuple, either
           (session, line, input) if output is False, or
           (session, line, (input, output)) if output is True.
         """
         if session <= 0:
             session += self.session_number
         if session==self.session_number:          # Current session
             return self._get_range_session(start, stop, raw, output)
         return super(HistoryManager, self).get_range(session, start, stop, raw,
                                                      output)
 
     ## ----------------------------
     ## Methods for storing history:
     ## ----------------------------
     def store_inputs(self, line_num, source, source_raw=None):
         """Store source and raw input in history and create input cache
         variables ``_i*``.
 
         Parameters
         ----------
         line_num : int
           The prompt number of this input.
 
         source : str
           Python input.
 
         source_raw : str, optional
           If given, this is the raw input without any IPython transformations
           applied to it.  If not given, ``source`` is used.
         """
         if source_raw is None:
             source_raw = source
         source = source.rstrip('\n')
         source_raw = source_raw.rstrip('\n')
 
         # do not store exit/quit commands
         if self._exit_re.match(source_raw.strip()):
             return
 
         self.input_hist_parsed.append(source)
         self.input_hist_raw.append(source_raw)
 
         with self.db_input_cache_lock:
             self.db_input_cache.append((line_num, source, source_raw))
             # Trigger to flush cache and write to DB.
             if len(self.db_input_cache) >= self.db_cache_size:
                 self.save_flag.set()
 
         # update the auto _i variables
         self._iii = self._ii
         self._ii = self._i
         self._i = self._i00
         self._i00 = source_raw
 
         # hackish access to user namespace to create _i1,_i2... dynamically
         new_i = '_i%s' % line_num
         to_main = {'_i': self._i,
                    '_ii': self._ii,
                    '_iii': self._iii,
                    new_i : self._i00 }
         
         if self.shell is not None:
             self.shell.push(to_main, interactive=False)
 
     def store_output(self, line_num):
         """If database output logging is enabled, this saves all the
         outputs from the indicated prompt number to the database. It's
         called by run_cell after code has been executed.
 
         Parameters
         ----------
         line_num : int
           The line number from which to save outputs
         """
         if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
             return
         output = self.output_hist_reprs[line_num]
 
         with self.db_output_cache_lock:
             self.db_output_cache.append((line_num, output))
         if self.db_cache_size <= 1:
             self.save_flag.set()
 
     def _writeout_input_cache(self, conn):
         with conn:
             for line in self.db_input_cache:
                 conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
                                 (self.session_number,)+line)
 
     def _writeout_output_cache(self, conn):
         with conn:
             for line in self.db_output_cache:
                 conn.execute("INSERT INTO output_history VALUES (?, ?, ?)",
                                 (self.session_number,)+line)
 
     @needs_sqlite
     def writeout_cache(self, conn=None):
         """Write any entries in the cache to the database."""
         if conn is None:
             conn = self.db
 
         with self.db_input_cache_lock:
             try:
                 self._writeout_input_cache(conn)
             except sqlite3.IntegrityError:
                 self.new_session(conn)
                 print("ERROR! Session/line number was not unique in",
                       "database. History logging moved to new session",
                                                 self.session_number)
                 try:
                     # Try writing to the new session. If this fails, don't
                     # recurse
                     self._writeout_input_cache(conn)
                 except sqlite3.IntegrityError:
                     pass
             finally:
                 self.db_input_cache = []
 
         with self.db_output_cache_lock:
             try:
                 self._writeout_output_cache(conn)
             except sqlite3.IntegrityError:
                 print("!! Session/line number for output was not unique",
                       "in database. Output will not be stored.")
             finally:
                 self.db_output_cache = []
 
 
 class HistorySavingThread(threading.Thread):
     """This thread takes care of writing history to the database, so that
     the UI isn't held up while that happens.
 
     It waits for the HistoryManager's save_flag to be set, then writes out
     the history cache. The main thread is responsible for setting the flag when
     the cache size reaches a defined threshold."""
     daemon = True
     stop_now = False
     enabled = True
     def __init__(self, history_manager):
         super(HistorySavingThread, self).__init__(name="IPythonHistorySavingThread")
         self.history_manager = history_manager
         self.enabled = history_manager.enabled
         atexit.register(self.stop)
 
     @needs_sqlite
     def run(self):
         # We need a separate db connection per thread:
         try:
             self.db = sqlite3.connect(self.history_manager.hist_file,
                             **self.history_manager.connection_options
             )
             while True:
                 self.history_manager.save_flag.wait()
                 if self.stop_now:
                     self.db.close()
                     return
                 self.history_manager.save_flag.clear()
                 self.history_manager.writeout_cache(self.db)
         except Exception as e:
             print(("The history saving thread hit an unexpected error (%s)."
                    "History will not be written to the database.") % repr(e))
 
     def stop(self):
         """This can be called from the main thread to safely stop this thread.
 
         Note that it does not attempt to write out remaining history before
         exiting. That should be done by calling the HistoryManager's
         end_session method."""
         self.stop_now = True
         self.history_manager.save_flag.set()
         self.join()
 
 
 # To match, e.g. ~5/8-~2/3
 range_re = re.compile(r"""
 ((?P<startsess>~?\d+)/)?
 (?P<start>\d+)?
 ((?P<sep>[\-:])
  ((?P<endsess>~?\d+)/)?
  (?P<end>\d+))?
 $""", re.VERBOSE)
 
 
 def extract_hist_ranges(ranges_str):
     """Turn a string of history ranges into 3-tuples of (session, start, stop).
 
     Examples
     --------
     >>> list(extract_hist_ranges("~8/5-~7/4 2"))
     [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
     """
     for range_str in ranges_str.split():
         rmatch = range_re.match(range_str)
         if not rmatch:
             continue
         start = rmatch.group("start")
         if start:
             start = int(start)
             end = rmatch.group("end")
             # If no end specified, get (a, a + 1)
             end = int(end) if end else start + 1
         else:  # start not specified
             if not rmatch.group('startsess'):  # no startsess
                 continue
             start = 1
             end = None  # provide the entire session hist
 
         if rmatch.group("sep") == "-":       # 1-3 == 1:4 --> [1, 2, 3]
             end += 1
         startsess = rmatch.group("startsess") or "0"
         endsess = rmatch.group("endsess") or startsess
         startsess = int(startsess.replace("~","-"))
         endsess = int(endsess.replace("~","-"))
         assert endsess >= startsess, "start session must be earlier than end session"
 
         if endsess == startsess:
             yield (startsess, start, end)
             continue
         # Multiple sessions in one range:
         yield (startsess, start, None)
         for sess in range(startsess+1, endsess):
             yield (sess, 1, None)
         yield (endsess, 1, end)
 
 
 def _format_lineno(session, line):
     """Helper function to format line numbers properly."""
     if session == 0:
         return str(line)
     return "%s#%s" % (session, line)

IPython/terminal/interactiveshell.py

@@ -1,543 +1,543
 """IPython terminal interface using prompt_toolkit"""
 from __future__ import print_function
 
 import os
 import sys
 import warnings
 from warnings import warn
 
 from IPython.core.interactiveshell import InteractiveShell, InteractiveShellABC
 from IPython.utils import io
 from IPython.utils.py3compat import PY3, cast_unicode_py2, input, string_types
 from IPython.utils.terminal import toggle_set_term_title, set_term_title
 from IPython.utils.process import abbrev_cwd
 from traitlets import Bool, Unicode, Dict, Integer, observe, Instance, Type, default, Enum, Union
 
 from prompt_toolkit.document import Document
 from prompt_toolkit.enums import DEFAULT_BUFFER, EditingMode
 from prompt_toolkit.filters import (HasFocus, Condition, IsDone)
 from prompt_toolkit.history import InMemoryHistory
 from prompt_toolkit.shortcuts import create_prompt_application, create_eventloop, create_prompt_layout, create_output
 from prompt_toolkit.interface import CommandLineInterface
 from prompt_toolkit.key_binding.manager import KeyBindingManager
 from prompt_toolkit.layout.processors import ConditionalProcessor, HighlightMatchingBracketProcessor
 from prompt_toolkit.styles import PygmentsStyle, DynamicStyle
 
 from pygments.styles import get_style_by_name, get_all_styles
 from pygments.style import Style
 from pygments.token import Token
 
 from .debugger import TerminalPdb, Pdb
 from .magics import TerminalMagics
 from .pt_inputhooks import get_inputhook_name_and_func
 from .prompts import Prompts, ClassicPrompts, RichPromptDisplayHook
 from .ptutils import IPythonPTCompleter, IPythonPTLexer
 from .shortcuts import register_ipython_shortcuts
 
 DISPLAY_BANNER_DEPRECATED = object()
 
 
 from pygments.style import Style
 
 class _NoStyle(Style): pass
 
 
 
 _style_overrides_light_bg = {
             Token.Prompt: '#0000ff',
             Token.PromptNum: '#0000ee bold',
             Token.OutPrompt: '#cc0000',
             Token.OutPromptNum: '#bb0000 bold',
 }
 
 _style_overrides_linux = {
             Token.Prompt: '#00cc00',
             Token.PromptNum: '#00bb00 bold',
             Token.OutPrompt: '#cc0000',
             Token.OutPromptNum: '#bb0000 bold',
 }
 
 
 
 def get_default_editor():
     try:
         ed = os.environ['EDITOR']
         if not PY3:
             ed = ed.decode()
         return ed
     except KeyError:
         pass
     except UnicodeError:
         warn("$EDITOR environment variable is not pure ASCII. Using platform "
              "default editor.")
 
     if os.name == 'posix':
         return 'vi'  # the only one guaranteed to be there!
     else:
         return 'notepad' # same in Windows!
 
 # conservatively check for tty
 # overridden streams can result in things like:
 # - sys.stdin = None
 # - no isatty method
 for _name in ('stdin', 'stdout', 'stderr'):
     _stream = getattr(sys, _name)
     if not _stream or not hasattr(_stream, 'isatty') or not _stream.isatty():
         _is_tty = False
         break
 else:
     _is_tty = True
 
 
 _use_simple_prompt = ('IPY_TEST_SIMPLE_PROMPT' in os.environ) or (not _is_tty)
 
 class TerminalInteractiveShell(InteractiveShell):
     space_for_menu = Integer(6, help='Number of line at the bottom of the screen '
                                                   'to reserve for the completion menu'
                             ).tag(config=True)
 
     def _space_for_menu_changed(self, old, new):
         self._update_layout()
 
     pt_cli = None
     debugger_history = None
     _pt_app = None
 
     simple_prompt = Bool(_use_simple_prompt,
         help="""Use `raw_input` for the REPL, without completion, multiline input, and prompt colors.
 
             Useful when controlling IPython as a subprocess, and piping STDIN/OUT/ERR. Known usage are:
             IPython own testing machinery, and emacs inferior-shell integration through elpy.
 
             This mode default to `True` if the `IPY_TEST_SIMPLE_PROMPT`
             environment variable is set, or the current terminal is not a tty.
 
             """
             ).tag(config=True)
 
     @property
     def debugger_cls(self):
         return Pdb if self.simple_prompt else TerminalPdb
 
     confirm_exit = Bool(True,
         help="""
         Set to confirm when you try to exit IPython with an EOF (Control-D
         in Unix, Control-Z/Enter in Windows). By typing 'exit' or 'quit',
         you can force a direct exit without any confirmation.""",
     ).tag(config=True)
 
     editing_mode = Unicode('emacs',
         help="Shortcut style to use at the prompt. 'vi' or 'emacs'.",
     ).tag(config=True)
 
     mouse_support = Bool(False,
         help="Enable mouse support in the prompt"
     ).tag(config=True)
 
     highlighting_style = Union([Unicode('legacy'), Type(klass=Style)],
         help="""The name or class of a Pygments style to use for syntax
         highlighting: \n %s""" % ', '.join(get_all_styles())
     ).tag(config=True)
 
 
     @observe('highlighting_style')
     @observe('colors')
     def _highlighting_style_changed(self, change):
         self.refresh_style()
 
     def refresh_style(self):
         self._style = self._make_style_from_name_or_cls(self.highlighting_style)
 
 
     highlighting_style_overrides = Dict(
         help="Override highlighting format for specific tokens"
     ).tag(config=True)
 
     true_color = Bool(False,
         help=("Use 24bit colors instead of 256 colors in prompt highlighting. "
               "If your terminal supports true color, the following command "
               "should print 'TRUECOLOR' in orange: "
               "printf \"\\x1b[38;2;255;100;0mTRUECOLOR\\x1b[0m\\n\"")
     ).tag(config=True)
 
     editor = Unicode(get_default_editor(),
         help="Set the editor used by IPython (default to $EDITOR/vi/notepad)."
     ).tag(config=True)
 
     prompts_class = Type(Prompts, help='Class used to generate Prompt token for prompt_toolkit').tag(config=True)
 
     prompts = Instance(Prompts)
 
     @default('prompts')
     def _prompts_default(self):
         return self.prompts_class(self)
 
     @observe('prompts')
     def _(self, change):
         self._update_layout()
 
     @default('displayhook_class')
     def _displayhook_class_default(self):
         return RichPromptDisplayHook
 
     term_title = Bool(True,
         help="Automatically set the terminal title"
     ).tag(config=True)
 
     display_completions = Enum(('column', 'multicolumn','readlinelike'),
         help= ( "Options for displaying tab completions, 'column', 'multicolumn', and "
                 "'readlinelike'. These options are for `prompt_toolkit`, see "
                 "`prompt_toolkit` documentation for more information."
                 ),
         default_value='multicolumn').tag(config=True)
 
     highlight_matching_brackets = Bool(True,
         help="Highlight matching brackets.",
     ).tag(config=True)
 
     extra_open_editor_shortcuts = Bool(False,
         help="Enable vi (v) or Emacs (C-X C-E) shortcuts to open an external editor. "
              "This is in addition to the F2 binding, which is always enabled."
     ).tag(config=True)
 
     @observe('term_title')
     def init_term_title(self, change=None):
         # Enable or disable the terminal title.
         if self.term_title:
             toggle_set_term_title(True)
             set_term_title('IPython: ' + abbrev_cwd())
         else:
             toggle_set_term_title(False)
 
     def init_display_formatter(self):
         super(TerminalInteractiveShell, self).init_display_formatter()
         # terminal only supports plain text
         self.display_formatter.active_types = ['text/plain']
         # disable `_ipython_display_`
         self.display_formatter.ipython_display_formatter.enabled = False
 
     def init_prompt_toolkit_cli(self):
         if self.simple_prompt:
             # Fall back to plain non-interactive output for tests.
             # This is very limited, and only accepts a single line.
             def prompt():
                 return cast_unicode_py2(input('In [%d]: ' % self.execution_count))
             self.prompt_for_code = prompt
             return
 
         # Set up keyboard shortcuts
         kbmanager = KeyBindingManager.for_prompt(
             enable_open_in_editor=self.extra_open_editor_shortcuts,
         )
         register_ipython_shortcuts(kbmanager.registry, self)
 
         # Pre-populate history from IPython's history database
         history = InMemoryHistory()
         last_cell = u""
         for __, ___, cell in self.history_manager.get_tail(self.history_load_length,
                                                         include_latest=True):
             # Ignore blank lines and consecutive duplicates
             cell = cell.rstrip()
             if cell and (cell != last_cell):
-                history.append(cast_unicode_py2(cell))
+                history.append(cell)
                 last_cell = cell
 
         self._style = self._make_style_from_name_or_cls(self.highlighting_style)
         style = DynamicStyle(lambda: self._style)
 
         editing_mode = getattr(EditingMode, self.editing_mode.upper())
 
         def patch_stdout(**kwargs):
             return self.pt_cli.patch_stdout_context(**kwargs)
 
         self._pt_app = create_prompt_application(
                             editing_mode=editing_mode,
                             key_bindings_registry=kbmanager.registry,
                             history=history,
                             completer=IPythonPTCompleter(shell=self,
                                                     patch_stdout=patch_stdout),
                             enable_history_search=True,
                             style=style,
                             mouse_support=self.mouse_support,
                             **self._layout_options()
         )
         self._eventloop = create_eventloop(self.inputhook)
         self.pt_cli = CommandLineInterface(
             self._pt_app, eventloop=self._eventloop,
             output=create_output(true_color=self.true_color))
 
     def _make_style_from_name_or_cls(self, name_or_cls):
         """
         Small wrapper that make an IPython compatible style from a style name
 
         We need that to add style for prompt ... etc.
         """
         style_overrides = {}
         if name_or_cls == 'legacy':
             legacy = self.colors.lower()
             if legacy == 'linux':
                 style_cls = get_style_by_name('monokai')
                 style_overrides = _style_overrides_linux
             elif legacy == 'lightbg':
                 style_overrides = _style_overrides_light_bg
                 style_cls = get_style_by_name('pastie')
             elif legacy == 'neutral':
                 # The default theme needs to be visible on both a dark background
                 # and a light background, because we can't tell what the terminal
                 # looks like. These tweaks to the default theme help with that.
                 style_cls = get_style_by_name('default')
                 style_overrides.update({
                     Token.Number: '#007700',
                     Token.Operator: 'noinherit',
                     Token.String: '#BB6622',
                     Token.Name.Function: '#2080D0',
                     Token.Name.Class: 'bold #2080D0',
                     Token.Name.Namespace: 'bold #2080D0',
                     Token.Prompt: '#009900',
                     Token.PromptNum: '#00ff00 bold',
                     Token.OutPrompt: '#990000',
                     Token.OutPromptNum: '#ff0000 bold',
                 })
 
                 # Hack: Due to limited color support on the Windows console
                 # the prompt colors will be wrong without this
                 if os.name == 'nt':
                     style_overrides.update({
                         Token.Prompt: '#ansidarkgreen',
                         Token.PromptNum: '#ansigreen bold',
                         Token.OutPrompt: '#ansidarkred',
                         Token.OutPromptNum: '#ansired bold',
                     })
             elif legacy =='nocolor':
                 style_cls=_NoStyle
                 style_overrides = {}
             else :
                 raise ValueError('Got unknown colors: ', legacy)
         else :
             if isinstance(name_or_cls, string_types):
                 style_cls = get_style_by_name(name_or_cls)
             else:
                 style_cls = name_or_cls
             style_overrides = {
                 Token.Prompt: '#009900',
                 Token.PromptNum: '#00ff00 bold',
                 Token.OutPrompt: '#990000',
                 Token.OutPromptNum: '#ff0000 bold',
             }
         style_overrides.update(self.highlighting_style_overrides)
         style = PygmentsStyle.from_defaults(pygments_style_cls=style_cls,
                                             style_dict=style_overrides)
 
         return style
 
     def _layout_options(self):
         """
         Return the current layout option for the current Terminal InteractiveShell
         """
         return {
                 'lexer':IPythonPTLexer(),
                 'reserve_space_for_menu':self.space_for_menu,
                 'get_prompt_tokens':self.prompts.in_prompt_tokens,
                 'get_continuation_tokens':self.prompts.continuation_prompt_tokens,
                 'multiline':True,
                 'display_completions_in_columns': (self.display_completions == 'multicolumn'),
 
                 # Highlight matching brackets, but only when this setting is
                 # enabled, and only when the DEFAULT_BUFFER has the focus.
                 'extra_input_processors': [ConditionalProcessor(
                         processor=HighlightMatchingBracketProcessor(chars='[](){}'),
                         filter=HasFocus(DEFAULT_BUFFER) & ~IsDone() &
                             Condition(lambda cli: self.highlight_matching_brackets))],
                 }
 
     def _update_layout(self):
         """
         Ask for a re computation of the application layout, if for example ,
         some configuration options have changed.
         """
         if self._pt_app:
             self._pt_app.layout = create_prompt_layout(**self._layout_options())
 
     def prompt_for_code(self):
         document = self.pt_cli.run(
             pre_run=self.pre_prompt, reset_current_buffer=True)
         return document.text
 
     def enable_win_unicode_console(self):
         if sys.version_info >= (3, 6):
             # Since PEP 528, Python uses the unicode APIs for the Windows
             # console by default, so WUC shouldn't be needed.
             return
 
         import win_unicode_console
 
         if PY3:
             win_unicode_console.enable()
         else:
             # https://github.com/ipython/ipython/issues/9768
             from win_unicode_console.streams import (TextStreamWrapper,
                                  stdout_text_transcoded, stderr_text_transcoded)
 
             class LenientStrStreamWrapper(TextStreamWrapper):
                 def write(self, s):
                     if isinstance(s, bytes):
                         s = s.decode(self.encoding, 'replace')
 
                     self.base.write(s)
 
             stdout_text_str = LenientStrStreamWrapper(stdout_text_transcoded)
             stderr_text_str = LenientStrStreamWrapper(stderr_text_transcoded)
 
             win_unicode_console.enable(stdout=stdout_text_str,
                                        stderr=stderr_text_str)
 
     def init_io(self):
         if sys.platform not in {'win32', 'cli'}:
             return
 
         self.enable_win_unicode_console()
 
         import colorama
         colorama.init()
 
         # For some reason we make these wrappers around stdout/stderr.
         # For now, we need to reset them so all output gets coloured.
         # https://github.com/ipython/ipython/issues/8669
         # io.std* are deprecated, but don't show our own deprecation warnings
         # during initialization of the deprecated API.
         with warnings.catch_warnings():
             warnings.simplefilter('ignore', DeprecationWarning)
             io.stdout = io.IOStream(sys.stdout)
             io.stderr = io.IOStream(sys.stderr)
 
     def init_magics(self):
         super(TerminalInteractiveShell, self).init_magics()
         self.register_magics(TerminalMagics)
 
     def init_alias(self):
         # The parent class defines aliases that can be safely used with any
         # frontend.
         super(TerminalInteractiveShell, self).init_alias()
 
         # Now define aliases that only make sense on the terminal, because they
         # need direct access to the console in a way that we can't emulate in
         # GUI or web frontend
         if os.name == 'posix':
             for cmd in ['clear', 'more', 'less', 'man']:
                 self.alias_manager.soft_define_alias(cmd, cmd)
 
 
     def __init__(self, *args, **kwargs):
         super(TerminalInteractiveShell, self).__init__(*args, **kwargs)
         self.init_prompt_toolkit_cli()
         self.init_term_title()
         self.keep_running = True
 
         self.debugger_history = InMemoryHistory()
 
     def ask_exit(self):
         self.keep_running = False
 
     rl_next_input = None
 
     def pre_prompt(self):
         if self.rl_next_input:
             # We can't set the buffer here, because it will be reset just after
             # this. Adding a callable to pre_run_callables does what we need
             # after the buffer is reset.
             s = cast_unicode_py2(self.rl_next_input)
             def set_doc():
                 self.pt_cli.application.buffer.document = Document(s)
             if hasattr(self.pt_cli, 'pre_run_callables'):
                 self.pt_cli.pre_run_callables.append(set_doc)
             else:
                 # Older version of prompt_toolkit; it's OK to set the document
                 # directly here.
                 set_doc()
             self.rl_next_input = None
 
     def interact(self, display_banner=DISPLAY_BANNER_DEPRECATED):
 
         if display_banner is not DISPLAY_BANNER_DEPRECATED:
             warn('interact `display_banner` argument is deprecated since IPython 5.0. Call `show_banner()` if needed.', DeprecationWarning, stacklevel=2)
 
         self.keep_running = True
         while self.keep_running:
             print(self.separate_in, end='')
 
             try:
                 code = self.prompt_for_code()
             except EOFError:
                 if (not self.confirm_exit) \
                         or self.ask_yes_no('Do you really want to exit ([y]/n)?','y','n'):
                     self.ask_exit()
 
             else:
                 if code:
                     self.run_cell(code, store_history=True)
 
     def mainloop(self, display_banner=DISPLAY_BANNER_DEPRECATED):
         # An extra layer of protection in case someone mashing Ctrl-C breaks
         # out of our internal code.
         if display_banner is not DISPLAY_BANNER_DEPRECATED:
             warn('mainloop `display_banner` argument is deprecated since IPython 5.0. Call `show_banner()` if needed.', DeprecationWarning, stacklevel=2)
         while True:
             try:
                 self.interact()
                 break
             except KeyboardInterrupt as e:
                 print("\n%s escaped interact()\n" % type(e).__name__)
             finally:
                 # An interrupt during the eventloop will mess up the
                 # internal state of the prompt_toolkit library.
                 # Stopping the eventloop fixes this, see
                 # https://github.com/ipython/ipython/pull/9867
                 if hasattr(self, '_eventloop'):
                     self._eventloop.stop()
 
     _inputhook = None
     def inputhook(self, context):
         if self._inputhook is not None:
             self._inputhook(context)
 
     active_eventloop = None
     def enable_gui(self, gui=None):
         if gui:
             self.active_eventloop, self._inputhook =\
                 get_inputhook_name_and_func(gui)
         else:
             self.active_eventloop = self._inputhook = None
 
     # Run !system commands directly, not through pipes, so terminal programs
     # work correctly.
     system = InteractiveShell.system_raw
 
     def auto_rewrite_input(self, cmd):
         """Overridden from the parent class to use fancy rewriting prompt"""
         if not self.show_rewritten_input:
             return
 
         tokens = self.prompts.rewrite_prompt_tokens()
         if self.pt_cli:
             self.pt_cli.print_tokens(tokens)
             print(cmd)
         else:
             prompt = ''.join(s for t, s in tokens)
             print(prompt, cmd, sep='')
 
     _prompts_before = None
     def switch_doctest_mode(self, mode):
         """Switch prompts to classic for %doctest_mode"""
         if mode:
             self._prompts_before = self.prompts
             self.prompts = ClassicPrompts(self)
         elif self._prompts_before:
             self.prompts = self._prompts_before
             self._prompts_before = None
         self._update_layout()
 
 
 InteractiveShellABC.register(TerminalInteractiveShell)
 
 if __name__ == '__main__':
     TerminalInteractiveShell.instance().interact()