IPython/core/historyapp.py

# 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()