Add command to trim the history database.

IPython/core/historyapp.py

@@ -0,0 +1,119 b''
+# encoding: utf-8
+"""
+An application for managing IPython history.
+
+To be invoked as the `ipython history` subcommand.
+"""
+from __future__ import print_function
+
+import os
+import sqlite3
+
+from IPython.config.application import Application
+from IPython.core.application import BaseIPythonApplication
+from IPython.utils.traitlets import Bool, Int, Dict
+
+trim_hist_help = """Trim the IPython history database to the last 1000 entries.
+
+This actually copies the last 1000 entries to a new database, and then replaces
+the old file with the new.
+"""
+
+class HistoryTrim(BaseIPythonApplication):
+    description = trim_hist_help
+    
+    backup = Bool(False, config=True,
+        help="Keep the old history file as history.sqlite.<N>")
+    
+    keep = Int(1000, config=True,
+        help="Number of recent lines to keep in the database.")
+    
+    flags = Dict(dict(
+        backup = ({'HistoryTrim' : {'backup' : True}},
+            "Set Application.log_level to 0, maximizing log output."
+        )
+    ))
+    
+    def start(self):
+        profile_dir = self.profile_dir.location
+        hist_file = os.path.join(profile_dir, 'history.sqlite')
+        con = sqlite3.connect(hist_file)
+
+        # Grab the recent history from the current database.
+        inputs = list(con.execute('SELECT session, line, source, source_raw FROM '
+                                'history ORDER BY session DESC, line DESC LIMIT ?', (self.keep+1,)))
+        if len(inputs) <= self.keep:
+            print("There are already at most %d entries in the history database." % self.keep)
+            print("Not doing anything.")
+            return
+        
+        print("Trimming history to the most recent %d entries." % self.keep)
+        
+        inputs.pop() # Remove the extra element we got to check the length.
+        inputs.reverse()
+        first_session = inputs[0][0]
+        outputs = list(con.execute('SELECT session, line, output FROM '
+                                   'output_history WHERE session >= ?', (first_session,)))
+        sessions = list(con.execute('SELECT session, start, end, num_cmds, remark FROM '
+                                    'sessions WHERE session >= ?', (first_session,)))
+        con.close()
+        
+        # Create the new history database.
+        new_hist_file = os.path.join(profile_dir, 'history.sqlite.new')
+        i = 0
+        while os.path.exists(new_hist_file):
+            # Make sure we don't interfere with an existing file.
+            i += 1
+            new_hist_file = os.path.join(profile_dir, 'history.sqlite.new'+str(i))
+        new_db = sqlite3.connect(new_hist_file)
+        new_db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
+                            primary key autoincrement, start timestamp,
+                            end timestamp, num_cmds integer, remark text)""")
+        new_db.execute("""CREATE TABLE IF NOT EXISTS history
+                        (session integer, line integer, source text, source_raw text,
+                        PRIMARY KEY (session, line))""")
+        new_db.execute("""CREATE TABLE IF NOT EXISTS output_history
+                        (session integer, line integer, output text,
+                        PRIMARY KEY (session, line))""")
+        new_db.commit()
+
+
+        with new_db:
+            # Add the recent history into the new database.
+            new_db.executemany('insert into sessions values (?,?,?,?,?)', sessions)
+            new_db.executemany('insert into history values (?,?,?,?)', inputs)
+            new_db.executemany('insert into output_history values (?,?,?)', outputs)
+        new_db.close()
+
+        if self.backup:
+            i = 1
+            backup_hist_file = os.path.join(profile_dir, 'history.sqlite.old.%d' % i)
+            while os.path.exists(backup_hist_file):
+                i += 1
+                backup_hist_file = os.path.join(profile_dir, 'history.sqlite.old.%d' % i)
+            os.rename(hist_file, backup_hist_file)
+            print("Backed up longer history file to", backup_hist_file)
+        else:
+            os.remove(hist_file)
+        
+        os.rename(new_hist_file, hist_file)
+
+
+class HistoryApp(Application):
+    name = u'ipython-history'
+    description = "Manage the IPython history database."
+
+    subcommands = Dict(dict(
+        trim = (HistoryTrim, HistoryTrim.description.splitlines()[0]),
+    ))
+
+    def start(self):
+        if self.subapp is None:
+            print("No subcommand specified. Must specify one of: %s" % \
+                                                    (self.subcommands.keys()))
+            print()
+            self.print_description()
+            self.print_subcommands()
+            self.exit(1)
+        else:
+            return self.subapp.start()

IPython/frontend/terminal/ipapp.py

@@ -241,6 +241,9 b' class TerminalIPythonApp(BaseIPythonApplication, InteractiveShellApp):'
         locate=('IPython.frontend.terminal.ipapp.LocateIPythonApp',
             LocateIPythonApp.description
         ),
+        history=('IPython.core.historyapp.HistoryApp',
+            "Manage the IPython history database."
+        ),
     ))
 
     # *do* autocreate requested profile, but don't create the config file.