1  # encoding: utf-8
   2  """
   3  An application for managing IPython history.
   4  
   5  To be invoked as the `ipython history` subcommand.
   6  """
   7  
   8  import sqlite3
   9  from pathlib import Path
  10  
  11  from traitlets.config.application import Application
  12  from .application import BaseIPythonApplication
  13  from traitlets import Bool, Int, Dict
  14  from ..utils.io import ask_yes_no
  15  
  16  trim_hist_help = """Trim the IPython history database to the last 1000 entries.
  17  
  18  This actually copies the last 1000 entries to a new database, and then replaces
  19  the old file with the new. Use the `--keep=` argument to specify a number
  20  other than 1000.
  21  """
  22  
  23  clear_hist_help = """Clear the IPython history database, deleting all entries.
  24  
  25  Because this is a destructive operation, IPython will prompt the user if they
  26  really want to do this. Passing a `-f` flag will force clearing without a
  27  prompt.
  28  
  29  This is an handy alias to `ipython history trim --keep=0`
  30  """
  31  
  32  
  33  class HistoryTrim(BaseIPythonApplication):
  34      description = trim_hist_help
  35  
  36      backup = Bool(False, help="Keep the old history file as history.sqlite.<N>").tag(
  37          config=True
  38      )
  39  
  40      keep = Int(1000, help="Number of recent lines to keep in the database.").tag(
  41          config=True
  42      )
  43  
  44      flags = Dict(  # type: ignore
  45          dict(backup=({"HistoryTrim": {"backup": True}}, backup.help))
  46      )
  47  
  48      aliases = Dict(dict(keep="HistoryTrim.keep"))  # type: ignore
  49  
  50      def start(self):
  51          profile_dir = Path(self.profile_dir.location)
  52          hist_file = profile_dir / "history.sqlite"
  53          con = sqlite3.connect(hist_file)
  54  
  55          # Grab the recent history from the current database.
  56          inputs = list(con.execute('SELECT session, line, source, source_raw FROM '
  57                                  'history ORDER BY session DESC, line DESC LIMIT ?', (self.keep+1,)))
  58          if len(inputs) <= self.keep:
  59              print("There are already at most %d entries in the history database." % self.keep)
  60              print("Not doing anything. Use --keep= argument to keep fewer entries")
  61              return
  62          
  63          print("Trimming history to the most recent %d entries." % self.keep)
  64          
  65          inputs.pop() # Remove the extra element we got to check the length.
  66          inputs.reverse()
  67          if inputs:
  68              first_session = inputs[0][0]
  69              outputs = list(con.execute('SELECT session, line, output FROM '
  70                                         'output_history WHERE session >= ?', (first_session,)))
  71              sessions = list(con.execute('SELECT session, start, end, num_cmds, remark FROM '
  72                                          'sessions WHERE session >= ?', (first_session,)))
  73          con.close()
  74          
  75          # Create the new history database.
  76          new_hist_file = profile_dir / "history.sqlite.new"
  77          i = 0
  78          while new_hist_file.exists():
  79              # Make sure we don't interfere with an existing file.
  80              i += 1
  81              new_hist_file = profile_dir / ("history.sqlite.new" + str(i))
  82          new_db = sqlite3.connect(new_hist_file)
  83          new_db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
  84                              primary key autoincrement, start timestamp,
  85                              end timestamp, num_cmds integer, remark text)""")
  86          new_db.execute("""CREATE TABLE IF NOT EXISTS history
  87                          (session integer, line integer, source text, source_raw text,
  88                          PRIMARY KEY (session, line))""")
  89          new_db.execute("""CREATE TABLE IF NOT EXISTS output_history
  90                          (session integer, line integer, output text,
  91                          PRIMARY KEY (session, line))""")
  92          new_db.commit()
  93  
  94  
  95          if inputs:
  96              with new_db:
  97                  # Add the recent history into the new database.
  98                  new_db.executemany('insert into sessions values (?,?,?,?,?)', sessions)
  99                  new_db.executemany('insert into history values (?,?,?,?)', inputs)
 100                  new_db.executemany('insert into output_history values (?,?,?)', outputs)
 101          new_db.close()
 102  
 103          if self.backup:
 104              i = 1
 105              backup_hist_file = profile_dir / ("history.sqlite.old.%d" % i)
 106              while backup_hist_file.exists():
 107                  i += 1
 108                  backup_hist_file = profile_dir / ("history.sqlite.old.%d" % i)
 109              hist_file.rename(backup_hist_file)
 110              print("Backed up longer history file to", backup_hist_file)
 111          else:
 112              hist_file.unlink()
 113  
 114          new_hist_file.rename(hist_file)
 115  
 116  
 117  class HistoryClear(HistoryTrim):
 118      description = clear_hist_help
 119      keep = Int(0, help="Number of recent lines to keep in the database.")
 120  
 121      force = Bool(False, help="Don't prompt user for confirmation").tag(config=True)
 122  
 123      flags = Dict(  # type: ignore
 124          dict(
 125              force=({"HistoryClear": {"force": True}}, force.help),
 126              f=({"HistoryTrim": {"force": True}}, force.help),
 127          )
 128      )
 129      aliases = Dict()  # type: ignore
 130  
 131      def start(self):
 132          if self.force or ask_yes_no(
 133              "Really delete all ipython history? ", default="no", interrupt="no"
 134          ):
 135              HistoryTrim.start(self)
 136  
 137  
 138  class HistoryApp(Application):
 139      name = "ipython-history"
 140      description = "Manage the IPython history database."
 141  
 142      subcommands = Dict(dict(
 143          trim = (HistoryTrim, HistoryTrim.description.splitlines()[0]),
 144          clear = (HistoryClear, HistoryClear.description.splitlines()[0]),
 145      ))
 146  
 147      def start(self):
 148          if self.subapp is None:
 149              print(
 150                  "No subcommand specified. Must specify one of: "
 151                  + ", ".join(map(repr, self.subcommands))
 152                  + ".\n"
 153              )
 154              self.print_description()
 155              self.print_subcommands()
 156              self.exit(1)
 157          else:
 158              return self.subapp.start()
 159