User Tools

Site Tools


python:database:sqlite:history_log

SQLite History Log

I needed a history log table showing old and new values of a table after insert, update and delete operations. I achieved to create history log for each DML operation by creating triggers on the table which I want to have history log and creating a new history log table to hold old and new values of each column. Simply, triggers create new records for each row inserted/updated/deleted. Instead of giving an example for this topic, I created and shared a TestCase using unittest to verify all operations are completed successfully.

Let's assume we have a table main as follows

CREATE TABLE IF NOT EXISTS "main" (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
name TEXT,
surname TEXT,
age INTEGER,
birthdate TIMESTAMP
);

Since we want to record all history of all columns, let's create a history table called main_history as follows. I added three more columns namely operation, user and timestamp to track which user do which operation when.

Since SQLite does not have multi user feature, I created a get_user function to fetch the user from application side and registered that function.

CREATE TABLE IF NOT EXISTS "main_history"
("id" INTEGER PRIMARY KEY ,
"old_id" INTEGER,
"new_id" INTEGER,
"old_name" TEXT,
"new_name" TEXT,
"old_surname" TEXT,
"new_surname" TEXT,
"old_age" INTEGER,
"new_age" INTEGER,
"old_birthdate" TIMESTAMP,
"new_birthdate" TIMESTAMP,
"operation" TEXT,
"user" TEXT DEFAULT (get_user()),
"timestamp" TIMESTAMP NOT NULL DEFAULT (datetime('now','localtime'))
);

The important part is to create the triggers so that history becomes available. Here are the triggers' statements

CREATE TRIGGER IF NOT EXISTS "main_ai" AFTER INSERT ON "main" FOR EACH ROW 
BEGIN
     INSERT INTO "main_history" (new_id, new_name,new_surname,new_age,new_birthdate, operation) VALUES
    (NEW.id, NEW.name, NEW.surname, NEW.age, NEW.birthdate,"INSERT");
END;
CREATE TRIGGER IF NOT EXISTS "main_bu" BEFORE UPDATE ON "main" FOR EACH ROW 
BEGIN
     INSERT INTO "main_history" (old_id, new_id, old_name, new_name,old_surname, new_surname,old_age, new_age,old_birthdate, new_birthdate, operation) VALUES
    (OLD.id, NEW.id, OLD.name, NEW.name, OLD.surname, NEW.surname, OLD.age, NEW.age, OLD.birthdate, NEW.birthdate, "UPDATE");
END;
CREATE TRIGGER IF NOT EXISTS "main_ad" AFTER DELETE ON "main" FOR EACH ROW 
BEGIN
     INSERT INTO "main_history" (old_id, old_name,old_surname,old_age,old_birthdate, operation) VALUES
    (OLD.id, OLD.name, OLD.surname,  OLD.age, OLD.birthdate, "DELETE");
END;

Triggers insert the records to main_history table appropriately as can be seen from the SQL code above.

Following code works under Python 2.7 and 3.x versions

test_trigger_history_log.py
# encoding: utf-8
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import absolute_import
 
import os
import sys
import sqlite3
import datetime
import unittest
 
PY2 = PY3 = False
if sys.version_info[0:2] == (2, 7):
    PY2 = True
elif sys.version_info[0] == 3:
    PY3 = True
else:
    raise Exception("Only Python 2.7 and 3.X are allowed")
 
if PY2:
    int_type = (int, long)
elif PY3:
    int_type = (int,)
 
 
def get_user():
    '''
    returns active user on application side
    '''
    return "test_user"
 
 
class SQLiteTriggerHistoryTestCase(unittest.TestCase):
 
    conn = None
    dbpath = ":memory:"  # ./historylog_example.sqlite"
 
    @classmethod
    def setUpClass(cls):
        # recreate db
        if cls.dbpath != ":memory:" and os.path.exists(cls.dbpath):
            os.remove(cls.dbpath)
        cls.conn = sqlite3.connect(cls.dbpath,
                                   isolation_level=None,
                                   detect_types=sqlite3.PARSE_DECLTYPES |
                                   sqlite3.PARSE_COLNAMES)
        cls.conn.row_factory = sqlite3.Row
        cls.conn.create_function("get_user", 0, get_user)
        create_stmt = '''
        CREATE TABLE IF NOT EXISTS "main" (
        id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
        name TEXT,
        surname TEXT,
        age INTEGER,
        birthdate TIMESTAMP
        );
        '''
        cls.conn.execute(create_stmt)
        create_stmt = '''
        CREATE TABLE IF NOT EXISTS "main_history"
        ("id" INTEGER PRIMARY KEY ,
        "old_id" INTEGER,
        "new_id" INTEGER,
        "old_name" TEXT,
        "new_name" TEXT,
        "old_surname" TEXT,
        "new_surname" TEXT,
        "old_age" INTEGER,
        "new_age" INTEGER,
        "old_birthdate" TIMESTAMP,
        "new_birthdate" TIMESTAMP,
        "operation" TEXT,
        "user" TEXT DEFAULT (get_user()),
        "timestamp" TIMESTAMP NOT NULL DEFAULT (datetime('now','localtime'))
        );
        '''
        cls.conn.execute(create_stmt)
        create_stmt = '''
        CREATE TRIGGER IF NOT EXISTS "main_ai" AFTER INSERT ON "main" FOR EACH ROW 
        BEGIN
             INSERT INTO "main_history" (new_id, new_name,new_surname,new_age,new_birthdate, operation) VALUES
            (NEW.id, NEW.name, NEW.surname, NEW.age, NEW.birthdate,"INSERT");
        END;
        '''
        cls.conn.execute(create_stmt)
        create_stmt = '''
        CREATE TRIGGER IF NOT EXISTS "main_bu" BEFORE UPDATE ON "main" FOR EACH ROW 
        BEGIN
             INSERT INTO "main_history" (old_id, new_id, old_name, new_name,old_surname, new_surname,old_age, new_age,old_birthdate, new_birthdate, operation) VALUES
            (OLD.id, NEW.id, OLD.name, NEW.name, OLD.surname, NEW.surname, OLD.age, NEW.age, OLD.birthdate, NEW.birthdate, "UPDATE");
        END;
        '''
        cls.conn.execute(create_stmt)
        create_stmt = '''
        CREATE TRIGGER IF NOT EXISTS "main_ad" AFTER DELETE ON "main" FOR EACH ROW 
        BEGIN
             INSERT INTO "main_history" (old_id, old_name,old_surname,old_age,old_birthdate, operation) VALUES
            (OLD.id, OLD.name, OLD.surname,  OLD.age, OLD.birthdate, "DELETE");
        END;
        '''
        cls.conn.execute(create_stmt)
 
    @classmethod
    def tearDownClass(cls):
        cls.conn.close()
 
    def setUp(self):
        self.cursor = self.conn.cursor()
        self.cursor.execute("DELETE FROM main")
        self.cursor.execute("DELETE FROM main_history")
        self.user = get_user()
 
    def tearDown(self):
        self.cursor.close()
        del self.cursor
        del self.user
 
    def test_insert(self):
        name = "Guido"
        surname = "Van Rossum"
        age = 59
        birthdate = datetime.datetime.strptime("1956-01-31", "%Y-%m-%d")
        self.cursor.execute('''INSERT INTO main (name,surname,age,birthdate)
                            VALUES (:name,:surname,:age,:birthdate)''',
                            {"name": name, "surname": surname, "age": age,
                             "birthdate": birthdate})
        self.assertIsInstance(self.cursor.lastrowid, int_type)
        lastrowid = self.cursor.lastrowid
        self.assertGreaterEqual(lastrowid, 0)
 
        self.cursor.execute("SELECT * FROM main")
        result = self.cursor.fetchall()
        self.assertGreater(len(result), 0)
 
        # verify insertion
        rec = dict(result[0])
        self.assertEqual(rec["id"], lastrowid)
        self.assertEqual(rec["name"], name)
        self.assertEqual(rec["surname"], surname)
        self.assertEqual(rec["age"], age)
        self.assertEqual(rec["birthdate"], birthdate)
 
        # verify history
        self.cursor.execute("SELECT * FROM main_history")
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
 
        # verify insertion
        rec = dict(result[0])
        self.assertIsNotNone(rec["id"])
        self.assertIsNone(rec["old_id"])
        self.assertEqual(rec["new_id"], lastrowid)
        self.assertIsNone(rec["old_name"])
        self.assertEqual(rec["new_name"], name)
        self.assertIsNone(rec["old_surname"])
        self.assertEqual(rec["new_surname"], surname)
        self.assertIsNone(rec["old_age"])
        self.assertEqual(rec["new_age"], age)
        self.assertIsNone(rec["old_birthdate"])
        self.assertEqual(rec["new_birthdate"], birthdate)
        self.assertEqual(rec["operation"], "INSERT")
        self.assertEqual(rec["user"], self.user)
        self.assertIsNotNone(rec["timestamp"])
 
    def test_update(self):
        name = "Guido"
        surname = "Van Rossum"
        age = 59
        birthdate = datetime.datetime.strptime("1956-01-31", "%Y-%m-%d")
        self.cursor.execute('''INSERT INTO main (name,surname,age,birthdate)
                            VALUES (:name,:surname,:age,:birthdate)''',
                            {"name": name, "surname": surname, "age": age,
                             "birthdate": birthdate})
        self.assertIsInstance(self.cursor.lastrowid, int_type)
        lastrowid = self.cursor.lastrowid
        self.assertGreaterEqual(lastrowid, 0)
 
        self.cursor.execute("SELECT * FROM main")
        result = self.cursor.fetchall()
        self.assertGreater(len(result), 0)
 
        # verify insertion
        rec = dict(result[0])
        self.assertEqual(rec["id"], lastrowid)
        self.assertEqual(rec["name"], name)
        self.assertEqual(rec["surname"], surname)
        self.assertEqual(rec["age"], age)
        self.assertEqual(rec["birthdate"], birthdate)
 
        # verify history
        self.cursor.execute("SELECT * FROM main_history")
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
 
        # verify insertion
        rec = dict(result[0])
        self.assertIsNotNone(rec["id"])
        self.assertIsNone(rec["old_id"])
        self.assertEqual(rec["new_id"], lastrowid)
        self.assertIsNone(rec["old_name"])
        self.assertEqual(rec["new_name"], name)
        self.assertIsNone(rec["old_surname"])
        self.assertEqual(rec["new_surname"], surname)
        self.assertIsNone(rec["old_age"])
        self.assertEqual(rec["new_age"], age)
        self.assertIsNone(rec["old_birthdate"])
        self.assertEqual(rec["new_birthdate"], birthdate)
        self.assertEqual(rec["operation"], "INSERT")
        self.assertEqual(rec["user"], self.user)
        self.assertIsNotNone(rec["timestamp"])
 
        new_age = 60
 
        self.cursor.execute('''UPDATE main SET age=:age WHERE id=:id''',
                            {"age": new_age, "id": lastrowid})
        # verify update
        self.cursor.execute('''SELECT * FROM main''')
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
        rec = dict(result[0])
        self.assertEqual(rec["age"], new_age)
        # verify update history
        self.cursor.execute('''SELECT * FROM main_history''')
        result = self.cursor.fetchall()
        # must be 2 records for insert and update
        self.assertEqual(len(result), 2)
        self.cursor.execute('''SELECT * FROM main_history
                            WHERE operation="UPDATE"''')
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
        rec = dict(result[0])
 
        self.assertIsNotNone(rec["id"])
        self.assertEqual(rec["old_id"], rec["new_id"])
        self.assertEqual(rec["old_name"], rec["new_name"])
        self.assertEqual(rec["old_surname"], rec["new_surname"])
        self.assertEqual(rec["new_age"], new_age)
        self.assertEqual(rec["old_age"], age)
        self.assertEqual(rec["old_birthdate"], rec["new_birthdate"])
        self.assertEqual(rec["user"], self.user)
        self.assertIsNotNone(rec["timestamp"])
 
    def test_delete(self):
        name = "Guido"
        surname = "Van Rossum"
        age = 60
        birthdate = datetime.datetime.strptime("1956-01-31", "%Y-%m-%d")
        self.cursor.execute('''INSERT INTO main (name,surname,age,birthdate)
                            VALUES (:name,:surname,:age,:birthdate)''',
                            {"name": name, "surname": surname, "age": age,
                             "birthdate": birthdate})
        self.assertIsInstance(self.cursor.lastrowid, int_type)
        lastrowid = self.cursor.lastrowid
        self.assertGreaterEqual(lastrowid, 0)
 
        self.cursor.execute("SELECT * FROM main")
        result = self.cursor.fetchall()
        self.assertGreater(len(result), 0)
 
        # verify insert on main table
        rec = dict(result[0])
        self.assertEqual(rec["id"], lastrowid)
        self.assertEqual(rec["name"], name)
        self.assertEqual(rec["surname"], surname)
        self.assertEqual(rec["age"], age)
        self.assertEqual(rec["birthdate"], birthdate)
 
        # verify history
        self.cursor.execute("SELECT * FROM main_history")
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
 
        # verify insert on main_history table
        rec = dict(result[0])
        self.assertIsNotNone(rec["id"])
        self.assertIsNone(rec["old_id"])
        self.assertEqual(rec["new_id"], lastrowid)
        self.assertIsNone(rec["old_name"])
        self.assertEqual(rec["new_name"], name)
        self.assertIsNone(rec["old_surname"])
        self.assertEqual(rec["new_surname"], surname)
        self.assertIsNone(rec["old_age"])
        self.assertEqual(rec["new_age"], age)
        self.assertIsNone(rec["old_birthdate"])
        self.assertEqual(rec["new_birthdate"], birthdate)
        self.assertEqual(rec["operation"], "INSERT")
        self.assertEqual(rec["user"], self.user)
        self.assertIsNotNone(rec["timestamp"])
 
 
        self.cursor.execute('''DELETE FROM main WHERE id=:id''',
                            {"id": lastrowid})
        # verify delete on main
        self.cursor.execute('''SELECT * FROM main''')
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 0)
 
        # must be 2 records for insert and delete
        self.cursor.execute('''SELECT * FROM main_history''')
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 2)
        self.cursor.execute('''SELECT * FROM main_history
                            WHERE operation="DELETE"''')
        result = self.cursor.fetchall()
        self.assertEqual(len(result), 1)
        rec = dict(result[0])
        self.assertIsNotNone(rec["id"])
        self.assertIsNone(rec["new_id"])
        self.assertIsNotNone(rec["old_id"])
        self.assertEqual(rec["old_name"], name)
        self.assertEqual(rec["old_surname"], surname)
        self.assertEqual(rec["old_age"], age)
        self.assertEqual(rec["old_birthdate"], birthdate)
        self.assertEqual(rec["operation"], "DELETE")
        self.assertIsNotNone(rec["timestamp"])
        self.assertIsNone(rec["new_name"])
        self.assertIsNone(rec["new_surname"])
        self.assertIsNone(rec["new_age"])
        self.assertIsNone(rec["new_birthdate"])
        self.assertEqual(rec["user"], self.user)
 
 
if __name__ == "__main__":
    unittest.main(verbosity=2)

Discussion

Enter your comment. Wiki syntax is allowed:
 
python/database/sqlite/history_log.txt · Last modified: 2016/08/08 09:41 by ozan