Last active
August 31, 2026 07:57
-
-
Save bockor/81cf26bbd7ca5acb3e0f99b8f61c5c02 to your computer and use it in GitHub Desktop.
A complete, dependency-free Python script (using only the standard library) that compares two SQLite databases.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| A complete, dependency-free Python script (using only the standard library) | |
| that compares two SQLite databases. | |
| It compares: | |
| ----------- | |
| Tables: Tables added or removed. | |
| Schemas: Columns added, removed, or if their data types changed. | |
| Data: Rows inserted, deleted, or updated. | |
| How it works: | |
| ------------- | |
| To accurately detect updates (rather than just seeing a row deletion and a row | |
| insertion), the script looks for a Primary Key in the table. | |
| If a table has no primary key, it falls back to comparing the raw row data, | |
| which will report changes as an insertion and a deletion. | |
| How to use it: | |
| -------------- | |
| Save the code as compare_sqlite.py. | |
| Open your terminal or command prompt. | |
| Run the script by providing the old database, the new database, and the | |
| name of the CSV file you want to create: | |
| "python compare_sqlite.py old_database.sqlite new_database.sqlite changes_report.csv" | |
| Understanding the Output CSV: | |
| ----------------------------- | |
| The CSV will contain 6 columns: | |
| Change Type: TABLE ADDED, TABLE REMOVED, COLUMN ADDED, COLUMN REMOVED, COLUMN TYPE CHANGED, ROW INSERTED, ROW DELETED, ROW UPDATED. | |
| Table Name: The table where the change occurred. | |
| Column / PK Field: If it's a schema change, this is the column name. | |
| If it's a data change, this is the name of the Primary Key column used to identify the row. | |
| Identifier (PK Value): The actual value of the Primary Key for the affected row (e.g., User ID 45). | |
| Old State / Value: For schema changes, the old data type. For data changes, a string representation of the entire old row tuple. | |
| New State / Value: For schema changes, the new data type. For data changes, a string representation of the entire new row tuple. | |
| Note: If a column is added to a table, the script will log the COLUMN ADDED, | |
| but it will also log a ROW UPDATED for every single row in that table. | |
| This is technically correct because the tuple structure of those rows changed | |
| (e.g., going from (1, "John") to (1, "John", NULL)). | |
| """ | |
| import sqlite3 | |
| import csv | |
| import argparse | |
| import sys | |
| def get_tables(conn): | |
| """Get a list of all user tables in the database.""" | |
| cursor = conn.execute( | |
| "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" | |
| ) | |
| return [row[0] for row in cursor.fetchall()] | |
| def get_schema(conn, table_name): | |
| """Get schema details as a dictionary keyed by column name.""" | |
| cursor = conn.execute(f"PRAGMA table_info('{table_name}')") | |
| return {row[1]: row for row in cursor.fetchall()} | |
| def get_primary_key(conn, table_name): | |
| """Find the primary key column for a table.""" | |
| cursor = conn.execute(f"PRAGMA table_info('{table_name}')") | |
| for row in cursor: | |
| if row[5] == 1: # pk column flag is 1 | |
| return row[1] | |
| return None | |
| def get_table_data(conn, table_name): | |
| """Fetch all rows and column names from a table.""" | |
| cursor = conn.execute(f"SELECT * FROM '{table_name}'") | |
| columns = [desc[0] for desc in cursor.description] | |
| return columns, cursor.fetchall() | |
| def compare_databases(db1_path, db2_path, output_csv): | |
| changes = [] | |
| try: | |
| conn1 = sqlite3.connect(db1_path) | |
| conn2 = sqlite3.connect(db2_path) | |
| except sqlite3.Error as e: | |
| print(f"Error connecting to databases: {e}") | |
| sys.exit(1) | |
| tables_db1 = set(get_tables(conn1)) | |
| tables_db2 = set(get_tables(conn2)) | |
| # 1. Check for Tables Added or Removed | |
| for table in tables_db2 - tables_db1: | |
| changes.append(('TABLE ADDED', table, '', '', '', '')) | |
| for table in tables_db1 - tables_db2: | |
| changes.append(('TABLE REMOVED', table, '', '', '', '')) | |
| # 2. Check Schema and Data for Common Tables | |
| common_tables = tables_db1.intersection(tables_db2) | |
| for table in common_tables: | |
| schema1 = get_schema(conn1, table) | |
| schema2 = get_schema(conn2, table) | |
| cols1 = set(schema1.keys()) | |
| cols2 = set(schema2.keys()) | |
| # Schema Differences | |
| for col in cols2 - cols1: | |
| changes.append(('COLUMN ADDED', table, col, '', '', f'Type: {schema2[col][2]}')) | |
| for col in cols1 - cols2: | |
| changes.append(('COLUMN REMOVED', table, col, '', '', f'Type: {schema1[col][2]}')) | |
| # Type changes for common columns | |
| for col in cols1.intersection(cols2): | |
| if schema1[col][2].upper() != schema2[col][2].upper(): | |
| changes.append(('COLUMN TYPE CHANGED', table, col, '', schema1[col][2], schema2[col][2])) | |
| # Data Differences | |
| pk_col = get_primary_key(conn1, table) | |
| cols_db1, data_db1 = get_table_data(conn1, table) | |
| cols_db2, data_db2 = get_table_data(conn2, table) | |
| if pk_col and pk_col in cols_db1 and pk_col in cols_db2: | |
| # If we have a Primary Key, we can accurately track Updates vs Inserts/Deletes | |
| pk_idx_1 = cols_db1.index(pk_col) | |
| pk_idx_2 = cols_db2.index(pk_col) | |
| # Map data by Primary Key | |
| dict_db1 = {row[pk_idx_1]: row for row in data_db1} | |
| dict_db2 = {row[pk_idx_2]: row for row in data_db2} | |
| pks_db1 = set(dict_db1.keys()) | |
| pks_db2 = set(dict_db2.keys()) | |
| # Deleted Rows | |
| for pk in pks_db1 - pks_db2: | |
| changes.append(('ROW DELETED', table, pk_col, pk, str(dict_db1[pk]), '')) | |
| # Inserted Rows | |
| for pk in pks_db2 - pks_db1: | |
| changes.append(('ROW INSERTED', table, pk_col, pk, '', str(dict_db2[pk]))) | |
| # Updated Rows | |
| for pk in pks_db1.intersection(pks_db2): | |
| # Note: If a new column was added, all rows will technically "update" to include NULL for that column | |
| if dict_db1[pk] != dict_db2[pk]: | |
| changes.append(('ROW UPDATED', table, pk_col, pk, str(dict_db1[pk]), str(dict_db2[pk]))) | |
| else: | |
| # Fallback if no Primary Key: Do a naive set comparison on row tuples | |
| set_db1 = set(data_db1) | |
| set_db2 = set(data_db2) | |
| for row in set_db2 - set_db1: | |
| changes.append(('ROW INSERTED (No PK)', table, '', '', '', str(row))) | |
| for row in set_db1 - set_db2: | |
| changes.append(('ROW DELETED (No PK)', table, '', '', str(row), '')) | |
| conn1.close() | |
| conn2.close() | |
| # Write results to CSV | |
| with open(output_csv, 'w', newline='', encoding='utf-8') as f: | |
| writer = csv.writer(f) | |
| writer.writerow([ | |
| 'Change Type', | |
| 'Table Name', | |
| 'Column / PK Field', | |
| 'Identifier (PK Value)', | |
| 'Old State / Value', | |
| 'New State / Value' | |
| ]) | |
| writer.writerows(changes) | |
| print(f"Comparison complete. Found {len(changes)} changes.") | |
| print(f"Results saved to: {output_csv}") | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description="Compare two SQLite databases and output changes to a CSV.") | |
| parser.add_argument("db1", help="Path to the original (baseline) SQLite database") | |
| parser.add_argument("db2", help="Path to the new (modified) SQLite database") | |
| parser.add_argument("output", help="Path for the output CSV file") | |
| args = parser.parse_args() | |
| compare_databases(args.db1, args.db2, args.output) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment