在数据库运维场景中,定期备份是保障数据安全的核心措施之一,使用Python编写脚本调用mysqldump工具实现自动备份,是很多中小规模项目的常用方案,这种方式无需依赖复杂的第三方库,适配性强且容易维护。

环境准备
在编写脚本之前,需要确认以下环境已经就绪:
- 服务器已安装Python3环境,且可以正常执行Python命令
- 已安装MySQL数据库,并且mysqldump命令可以在终端直接调用,可通过
mysqldump --version验证是否安装成功 - 拥有待备份数据库的相关权限,包括读取数据库数据、导出数据的权限
核心实现逻辑
整个备份流程的核心是通过subprocess模块执行mysqldump命令,将导出的数据保存到指定文件中,同时需要处理命令执行过程中的异常,确保备份失败时能被及时发现。
基础备份脚本实现
以下是一个简单的单数据库备份脚本示例:
import subprocess
import os
import time
from datetime import datetime
# 数据库配置信息
DB_HOST = "127.0.0.1"
DB_PORT = "3306"
DB_USER = "root"
DB_PASSWORD = "your_password"
DB_NAME = "test_db"
# 备份文件存储目录
BACKUP_DIR = "/data/db_backup"
def backup_database():
# 确保备份目录存在
if not os.path.exists(BACKUP_DIR):
os.makedirs(BACKUP_DIR)
# 生成带时间戳的备份文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_file = os.path.join(BACKUP_DIR, f"{DB_NAME}_{timestamp}.sql")
# 拼接mysqldump命令
cmd = [
"mysqldump",
f"-h{DB_HOST}",
f"-P{DB_PORT}",
f"-u{DB_USER}",
f"-p{DB_PASSWORD}",
DB_NAME,
"-r",
backup_file
]
try:
# 执行命令,捕获输出和错误
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if result.returncode == 0:
print(f"数据库备份成功,文件路径:{backup_file}")
return True
else:
print(f"数据库备份失败,错误信息:{result.stderr}")
return False
except Exception as e:
print(f"执行备份命令时出现异常:{str(e)}")
return False
if __name__ == "__main__":
backup_database()
代码说明
上述代码中,首先定义了数据库的连接信息和备份目录,然后生成带时间戳的备份文件名避免文件覆盖。通过列表形式拼接mysqldump命令,避免命令注入风险,使用subprocess.run执行命令并获取返回结果,根据返回码判断备份是否成功。
优化备份脚本
基础脚本可以满足简单场景需求,实际使用中还可以做以下优化:
支持多数据库备份
如果需要同时备份多个数据库,可以修改脚本支持数据库列表配置:
import subprocess
import os
from datetime import datetime
DB_HOST = "127.0.0.1"
DB_PORT = "3306"
DB_USER = "root"
DB_PASSWORD = "your_password"
# 待备份的数据库列表
DB_LIST = ["test_db1", "test_db2", "test_db3"]
BACKUP_DIR = "/data/db_backup"
def backup_multi_database():
if not os.path.exists(BACKUP_DIR):
os.makedirs(BACKUP_DIR)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
for db_name in DB_LIST:
backup_file = os.path.join(BACKUP_DIR, f"{db_name}_{timestamp}.sql")
cmd = [
"mysqldump",
f"-h{DB_HOST}",
f"-P{DB_PORT}",
f"-u{DB_USER}",
f"-p{DB_PASSWORD}",
db_name,
"-r",
backup_file
]
try:
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if result.returncode == 0:
print(f"数据库{db_name}备份成功")
else:
print(f"数据库{db_name}备份失败,错误:{result.stderr}")
except Exception as e:
print(f"备份{db_name}时出现异常:{str(e)}")
if __name__ == "__main__":
backup_multi_database()
添加备份文件清理逻辑
为了避免备份文件占用过多磁盘空间,可以添加自动清理过期备份的逻辑:
import os
import time
def clean_old_backup(backup_dir, keep_days=7):
# 计算过期时间戳
expire_time = time.time() - keep_days * 24 * 3600
for root, dirs, files in os.walk(backup_dir):
for file in files:
file_path = os.path.join(root, file)
# 获取文件修改时间
file_mtime = os.path.getmtime(file_path)
if file_mtime < expire_time:
os.remove(file_path)
print(f"删除过期备份文件:{file_path}")
# 在备份完成后调用清理函数
# clean_old_backup(BACKUP_DIR, keep_days=7)
配置定时执行
脚本编写完成后,可以通过系统的定时任务工具实现自动执行,Linux系统可以使用crontab配置:
执行crontab -e命令,添加如下配置,实现每天凌晨2点执行备份脚本:
0 2 * * * /usr/bin/python3 /path/to/backup_script.py >> /var/log/db_backup.log 2>&1
Windows系统可以使用任务计划程序,创建基本任务,设置触发器和执行Python脚本的操作即可。
注意事项
- 数据库密码直接写在脚本中存在安全风险,生产环境建议使用配置文件存储密码,或者对密码进行加密处理
- mysqldump命令的
-p参数和密码之间没有空格,如果密码包含特殊字符,可能需要使用引号包裹密码 - 如果数据库数据量较大,备份过程可能会占用较多系统资源,可以选择业务低峰期执行备份任务
- 建议定期验证备份文件的可用性,避免备份文件损坏导致无法恢复数据
Pythonsubprocessmysqldump数据库备份修改时间:2026-07-23 13:48:29