如何通过Python客户端实现类似mysqldump的MySQL数据库备份？

1. 纯Python手动实现mysqldump逻辑 #

利用你正在使用的mysql.connector，直接通过SQL语句导出表结构和数据，完全不需要依赖客户端MySQL工具：

• 导出表结构：通过SHOW CREATE TABLE获取每个表的创建语句，写入备份文件
• 导出表数据：用SELECT *查询全表数据，转换为INSERT语句或CSV格式（CSV更易读，恢复也灵活）

示例代码：

import mysql.connector
from mysql.connector import Error

def backup_database(host, user, password, db_name, backup_path):
    try:
        conn = mysql.connector.connect(host=host, user=user, password=password, database=db_name)
        cursor = conn.cursor()

        # 获取所有表名
        cursor.execute("SHOW TABLES")
        tables = [tbl[0] for tbl in cursor.fetchall()]

        with open(backup_path, 'w', encoding='utf-8') as f:
            # 写入数据库创建语句（可选）
            cursor.execute(f"SHOW CREATE DATABASE {db_name}")
            db_create_sql = cursor.fetchone()[1]
            f.write(f"{db_create_sql};\n\n")

            for table in tables:
                # 写入表结构
                cursor.execute(f"SHOW CREATE TABLE {table}")
                table_create_sql = cursor.fetchone()[1]
                f.write(f"{table_create_sql};\n\n")

                # 写入表数据（INSERT格式）
                cursor.execute(f"SELECT * FROM {table}")
                rows = cursor.fetchall()
                if not rows:
                    continue

                # 获取列名
                cursor.execute(f"DESCRIBE {table}")
                columns = [col[0] for col in cursor.fetchall()]
                col_str = ', '.join(columns)
                f.write(f"INSERT INTO {table} ({col_str}) VALUES\n")

                for idx, row in enumerate(rows):
                    # 处理字符串转义，避免SQL语法错误
                    escaped_vals = []
                    for val in row:
                        if isinstance(val, str):
                            escaped_vals.append(f"'{val.replace(\"'\", \"''\")}'")
                        elif val is None:
                            escaped_vals.append("NULL")
                        else:
                            escaped_vals.append(str(val))
                    row_str = '(' + ', '.join(escaped_vals) + ')'
                    if idx == len(rows) - 1:
                        f.write(f"{row_str};\n\n")
                    else:
                        f.write(f"{row_str},\n")

        print(f"备份完成，文件路径：{backup_path}")
    except Error as e:
        print(f"备份失败：{str(e)}")
    finally:
        if conn.is_connected():
            cursor.close()
            conn.close()

# 调用示例
backup_database("服务器地址", "用户名", "密码", "数据库名", "./db_backup.sql")

2. 用第三方Python库简化操作 #

有纯Python实现的备份库，无需依赖系统mysqldump：

• mysqldump-python：完全模拟mysqldump功能，安装后直接调用
  安装命令：

  pip install mysqldump-python
  

  使用示例：

  from mysqldump import MySQLDump
  
  md = MySQLDump(
      host='服务器地址',
      user='用户名',
      password='密码',
      database='数据库名'
  )
  md.dump('./db_backup.sql')
  

• sqlalchemy + pandas：适合导出为CSV/Excel格式（方便后续数据分析或恢复）

  import pandas as pd
  from sqlalchemy import create_engine
  
  # 建立连接
  engine = create_engine('mysql+mysqlconnector://用户名:密码@服务器地址/数据库名')
  conn = engine.connect()
  
  # 获取所有表名
  tables = pd.read_sql("SHOW TABLES", conn)[conn.execute("SHOW TABLES").keys()[0]].tolist()
  
  # 逐个表导出为CSV
  for table in tables:
      df = pd.read_sql(f"SELECT * FROM {table}", conn)
      df.to_csv(f"./{table}_backup.csv", index=False, encoding='utf-8')
  
  conn.close()
  

3. 增量备份优化（适合大数据量场景） #

如果全量备份耗时太长，可以只备份每次提交的增量数据：

• 客户端提交数据时，同步记录本次操作的SQL语句（INSERT/UPDATE/DELETE）到本地备份文件
• 定期执行全量备份，平时保留增量操作记录；恢复时先导入全量备份，再依次执行增量SQL

关键注意事项 #

• 备份文件建议加密存储，防止数据泄露
• 可将备份文件上传至公司内部云存储，进一步提升安全性
• 定期测试备份恢复流程，确保备份文件有效

内容的提问来源于stack exchange，提问作者d'Elbreil Clément