compare_databases.py

#!/usr/bin/env python3
# 1. Generate testdb1 using origin/develop
#    `wbia-init-testdbs -r`
# 2. Back up the directory
#    `mv var/data/testdb1 var/data/testdb1.orig`
# 3. Generate testdb1 using origin/mmulich/sqlalchemy-queries
#    `wbia-init-testdbs -r`
# 4. Run this script to find the differences in the sqlite databases:
#    `./compare_databases.py var/data/testdb1.orig/ var/data/testdb1/`
import difflib
import subprocess
import sys

import sqlite3


def get_tables(filename):
    p1 = subprocess.Popen(['echo', '.tables'], stdout=subprocess.PIPE)
    p2 = subprocess.Popen(
        ['sqlite3', filename], stdin=p1.stdout, stdout=subprocess.PIPE)
    p1.stdout.close()
    return p2.communicate()[0].decode('utf-8').split()


def get_table_contents(filename, tables):
    with sqlite3.connect(filename) as conn:
        cursor = conn.cursor()
        for table in tables:
            cursor.execute(f'select * from {table}')
            yield cursor.fetchall()


def compare_databases(db1, db2):
    tables1 = get_tables(db1)
    tables2 = get_tables(db2)
    db1_contents = dict(zip(tables1, list(get_table_contents(db1, tables1))))
    db2_contents = dict(zip(tables2, list(get_table_contents(db2, tables2))))
    tables = []
    for table in tables1:
        table_content1 = [f'{a}\n' for a in db1_contents[table]]
        table_content2 = [f'{a}\n' for a in db2_contents[table]]
        if table_content1 != table_content2:
            print(table)
            print(''.join(list(difflib.unified_diff(
                table_content1, table_content2))))
            tables.append(table)
    return tables


def get_databases(dir_path):
    return sorted(subprocess.check_output(
        ['find', dir_path, '-name', '*.sqlite*']).decode(
        'utf-8').splitlines())


if __name__ == '__main__':
    if len(sys.argv) != 3:
        sys.stderr.write(f'Usage: {sys.argv[0]} <path-to-db1> <path-to-db2>\n')
        sys.exit(1)

    path1, path2 = sys.argv[1:]
    exit_code = 0
    for db1, db2 in zip(get_databases(path1), get_databases(path2)):
        print(f'Comparing {db1} and {db2}...')
        if compare_databases(db1, db2):
            exit_code = 1
    sys.exit(exit_code)