如何将MySQL数据库迁移至SQLite？实现一键迁移按钮方法

Hey there! Let's break down how to solve your problem step by step—first the core data migration logic, then wrapping it in a one-click button for ease of use. I'll cover both desktop and web-based options since you didn't specify a platform.

一、核心迁移逻辑：MySQL → SQLite #

The biggest hurdle here is handling syntax and data type differences between the two databases. We'll use Python (since it's cross-platform and has great DB libraries) to read from MySQL, convert schema/data, and write to SQLite.

Step 1: Setup Dependencies #

First install the required libraries:

pip install pymysql sqlite3  # sqlite3 is usually included with Python

Step 2: Migration Code (Core Logic) #

Here's a reusable function that handles schema conversion, data reading, and bulk insertion—with error handling and transaction support:

import pymysql
import sqlite3

# Configure your MySQL and SQLite paths here
MYSQL_CONFIG = {
    "host": "localhost",
    "user": "your_mysql_username",
    "password": "your_mysql_password",
    "database": "your_mysql_db_name",
    "charset": "utf8mb4"
}
SQLITE_DB_PATH = "./target_sqlite.db"

# Map MySQL data types to SQLite equivalents
TYPE_MAPPING = {
    "int": "INTEGER",
    "tinyint": "INTEGER",
    "bigint": "INTEGER",
    "float": "REAL",
    "double": "REAL",
    "decimal": "REAL",
    "char": "TEXT",
    "varchar": "TEXT",
    "text": "TEXT",
    "longtext": "TEXT",
    "date": "TEXT",
    "datetime": "TEXT",
    "enum": "TEXT",  # SQLite doesn't support ENUM, fall back to TEXT
    "blob": "BLOB"
}

def migrate_data():
    try:
        # Connect to MySQL and SQLite
        mysql_conn = pymysql.connect(**MYSQL_CONFIG)
        mysql_cursor = mysql_conn.cursor()
        
        sqlite_conn = sqlite3.connect(SQLITE_DB_PATH)
        sqlite_cursor = sqlite_conn.cursor()
        sqlite_cursor.execute("PRAGMA foreign_keys = ON;")  # Enable foreign keys in SQLite

        # Get all table names from MySQL
        mysql_cursor.execute("SHOW TABLES;")
        tables = [table[0] for table in mysql_cursor.fetchall()]

        for table in tables:
            # Get MySQL table schema
            mysql_cursor.execute(f"DESCRIBE {table};")
            columns = mysql_cursor.fetchall()

            # Build SQLite CREATE TABLE statement
            create_sql = f"CREATE TABLE IF NOT EXISTS `{table}` ("
            column_defs = []
            primary_keys = []

            for col in columns:
                col_name, col_type_raw, col_null, col_key, col_default, col_extra = col
                col_type = col_type_raw.split("(")[0].lower()  # Strip size (e.g., varchar(255) → varchar)
                sqlite_type = TYPE_MAPPING.get(col_type, "TEXT")

                # Handle primary keys and auto-increment
                if col_key == "PRI":
                    primary_keys.append(col_name)
                    if col_extra == "auto_increment":
                        column_defs.append(f"`{col_name}` {sqlite_type} PRIMARY KEY AUTOINCREMENT")
                    else:
                        column_defs.append(f"`{col_name}` {sqlite_type} PRIMARY KEY")
                else:
                    # Handle NOT NULL and default values
                    def_part = ""
                    if col_null == "NO":
                        def_part += " NOT NULL"
                    if col_default is not None:
                        def_part += f" DEFAULT '{col_default}'" if sqlite_type in ["TEXT", "BLOB"] else f" DEFAULT {col_default}"
                    column_defs.append(f"`{col_name}` {sqlite_type}{def_part}")

            # Add composite primary keys if needed
            if len(primary_keys) > 1:
                column_defs.append(f"PRIMARY KEY ({', '.join(primary_keys)})")

            create_sql += ", ".join(column_defs) + ");"
            sqlite_cursor.execute(create_sql)

            # Bulk insert data from MySQL to SQLite
            mysql_cursor.execute(f"SELECT * FROM {table};")
            rows = mysql_cursor.fetchall()
            if rows:
                placeholders = ", ".join(["?"] * len(columns))
                insert_sql = f"INSERT INTO `{table}` VALUES ({placeholders});"
                sqlite_cursor.executemany(insert_sql, rows)

        # Commit changes and clean up
        sqlite_conn.commit()
        print("Migration completed successfully!")

    except Exception as e:
        # Rollback on error to avoid partial data
        if "sqlite_conn" in locals():
            sqlite_conn.rollback()
        print(f"Migration failed: {str(e)}")
    finally:
        # Close all connections
        for conn in [mysql_conn, sqlite_conn]:
            if conn:
                conn.close()

二、一键导入按钮实现 #

Now let's wrap the migration logic in a button for easy access. I'll cover two common scenarios:

Option 1: Desktop App (Tkinter) #

Perfect for local use without a web server:

from tkinter import Tk, Button, messagebox

def create_desktop_gui():
    root = Tk()
    root.title("MySQL → SQLite Migrator")
    root.geometry("350x120")

    # Button that triggers migration
    migrate_btn = Button(
        root,
        text="Start Migration",
        command=lambda: migrate_data() or messagebox.showinfo("Success", "Data migrated!"),
        width=20,
        height=2,
        bg="#4CAF50",
        fg="white"
    )
    migrate_btn.pack(pady=30)

    root.mainloop()

if __name__ == "__main__":
    create_desktop_gui()

Option 2: Web App (Flask) #

Great if you need to access the migrator from a browser:

from flask import Flask, render_template_string, redirect, url_for

app = Flask(__name__)

@app.route("/")
def home():
    return render_template_string("""
        <style>
            body { font-family: Arial; max-width: 600px; margin: 2rem auto; padding: 0 1rem; }
            button { padding: 0.8rem 2rem; background: #4CAF50; color: white; border: none; border-radius: 4px; cursor: pointer; }
        </style>
        <h1>MySQL to SQLite One-Click Migration</h1>
        <form method="POST" action="{{ url_for('run_migration') }}">
            <button type="submit">Start Migration</button>
        </form>
    """)

@app.route("/migrate", methods=["POST"])
def run_migration():
    try:
        migrate_data()
        return "<h2>Migration Successful!</h2><a href='/'>Go back</a>"
    except Exception as e:
        return f"<h2>Migration Failed:</h2><p>{str(e)}</p><a href='/'>Go back</a>"

if __name__ == "__main__":
    app.run(debug=True)

Key Notes to Avoid Issues #

• Test First: Run the migration on a copy of your MySQL database before using production data.
• Data Type Edge Cases: For rare MySQL types not in the mapping, adjust the TYPE_MAPPING dictionary.
• Large Datasets: If you have huge tables, modify the code to fetch data in batches (e.g., LIMIT 1000 in MySQL queries) to avoid memory issues.
• Foreign Keys: SQLite disables foreign keys by default—we enabled them with PRAGMA foreign_keys = ON;, but double-check your relationships.

内容的提问来源于stack exchange，提问作者Abuzar Ahmadi