migrate_sqlite_to_mysql.py

# -----------------------------------------------------------
# ✅ MySQL Connection Test
# -----------------------------------------------------------
# This section is used to test if the MySQL database connection 
# is working correctly before running the migration.
# To test:
# 1. Ensure MySQL is running (e.g., via XAMPP or standalone server).
# 2. Update 'config.ini' with correct MySQL credentials.
# 3. Uncomment the section below and run:
#    python migrate_sqlite_to_mysql.py
# 4. If successful, it will print: "Connected successfully"
# -----------------------------------------------------------

# Uncomment the section below to test MySQL connection
# config = configparser.ConfigParser()
# config.read('config.ini')
# 
# mysql_config = {
#     'host': config['mysql']['host'],
#     'user': config['mysql']['user'],
#     'password': config['mysql']['password'],
#     'database': config['mysql']['database']
# }
# 
# try:
#     conn = mysql.connector.connect(**mysql_config)
#     print("Connected successfully")
#     conn.close()
# except mysql.connector.Error as e:
#     print(f"Error connecting to MySQL: {e}")

# -----------------------------------------------------------
# ✅ SQLite to MySQL Migration Script
# -----------------------------------------------------------
# This script automates the migration of data from an SQLite
# database to a MySQL database, ensuring smooth conversion of
# data types and table dependencies.
# -----------------------------------------------------------



# # SQLITE to MySQL

import sqlite3
import mysql.connector
import configparser
from mysql.connector import Error
from typing import List, Tuple
import logging

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def get_table_order(sqlite_cursor) -> List[str]:
    """Get tables in the correct order considering dependencies."""
    sqlite_cursor.execute("""
        SELECT name FROM sqlite_master
        WHERE type='table'
        ORDER BY CASE
            WHEN name = 'django_migrations' THEN 1
            WHEN name = 'auth_group' THEN 2
            WHEN name = 'auth_user' THEN 3
            WHEN name = 'auth_permission' THEN 4
            ELSE 5
        END, name
    """)
    return [table[0] for table in sqlite_cursor.fetchall() if table[0] != 'sqlite_sequence']

def get_column_types(sqlite_cursor, table_name: str) -> List[Tuple[str, str]]:
    """Get column names and their SQLite types."""
    sqlite_cursor.execute(f"PRAGMA table_info({table_name})")
    return [(col[1], col[2].upper()) for col in sqlite_cursor.fetchall()]

def convert_sqlite_value(value, sqlite_type: str):
    """Convert SQLite values to MySQL compatible format."""
    if value is None:
        return None
    # Handle BOOLEAN type
    if sqlite_type == 'BOOLEAN':
        return 1 if value else 0
    # Handle DATETIME
    if 'DATETIME' in sqlite_type and isinstance(value, str):
        return value.replace('T', ' ').split('.')[0]
    return value

def transfer_data(sqlite_file: str, mysql_config: dict):
    """Transfer data from SQLite to MySQL database."""
    sqlite_conn = None
    mysql_conn = None
    
    try:
        # Connect to SQLite
        sqlite_conn = sqlite3.connect(sqlite_file)
        sqlite_cursor = sqlite_conn.cursor()
        
        # Connect to MySQL
        mysql_conn = mysql.connector.connect(**mysql_config)
        mysql_cursor = mysql_conn.cursor(buffered=True)
        
        # Disable foreign key checks and autocommit
        mysql_cursor.execute("SET foreign_key_checks = 0;")
        mysql_conn.autocommit = False
        
        # Get tables in the correct order
        tables = get_table_order(sqlite_cursor)
        
        for table_name in tables:
            logging.info(f"Processing table: {table_name}")
            
            # Get column information
            column_info = get_column_types(sqlite_cursor, table_name)
            column_names = [col[0] for col in column_info]
            
            # Clear existing data
            mysql_cursor.execute(f"TRUNCATE TABLE `{table_name}`;")
            
            # Get data from SQLite
            sqlite_cursor.execute(f"SELECT * FROM {table_name}")
            rows = sqlite_cursor.fetchall()
            
            if rows:
                # Prepare MySQL query
                columns_string = ", ".join([f"`{col}`" for col in column_names])
                placeholders = ", ".join(["%s"] * len(column_names))
                replace_query = f"REPLACE INTO `{table_name}` ({columns_string}) VALUES ({placeholders})"
                
                # Process and insert data in batches
                batch_size = 1000
                for i in range(0, len(rows), batch_size):
                    batch = rows[i:i + batch_size]
                    # Convert data types
                    converted_batch = [
                        [convert_sqlite_value(value, column_info[j][1])
                         for j, value in enumerate(row)]
                        for row in batch
                    ]
                    
                    try:
                        mysql_cursor.executemany(replace_query, converted_batch)
                        mysql_conn.commit()
                        logging.info(f"Inserted {len(converted_batch)} rows into {table_name}")
                    except mysql.connector.Error as e:
                        mysql_conn.rollback()
                        logging.error(f"Error inserting into {table_name}: {e}")
                        raise
        
        # Re-enable foreign key checks
        mysql_cursor.execute("SET foreign_key_checks = 1;")
        logging.info("Data transfer completed successfully")
        
    except sqlite3.Error as e:
        logging.error(f"SQLite Error: {e}")
        raise
    except mysql.connector.Error as e:
        logging.error(f"MySQL Error: {e}")
        raise
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        raise
    finally:
        # Clean up connections
        if mysql_conn and mysql_conn.is_connected():
            mysql_cursor.close()
            mysql_conn.close()
        if sqlite_conn:
            sqlite_conn.close()

if __name__ == "__main__":
    # Load configuration
    config = configparser.ConfigParser()
    config.read('config.ini')

    sqlite_file = config['sqlite']['database']
    mysql_config = {
        'host': config['mysql']['host'],
        'user': config['mysql']['user'],
        'password': config['mysql']['password'],
        'database': config['mysql']['database'],
        'charset': config['mysql']['charset']
    }
    
    try:
        transfer_data(sqlite_file, mysql_config)
    except Exception as e:
        logging.error(f"Migration failed: {e}")