python实现mysql数据库自动备份
最新回答
温茶
2020-06-09 04:51:09
以下是使用Python实现MySQL数据库自动备份的完整方案，包含代码优化和关键注意事项：
优化后的备份脚本import osimport timeimport pymysqlfrom datetime import datetimeimport loggingfrom pathlib import Path# 配置日志记录logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('db_backup.log'), logging.StreamHandler() ])class MySQLBackup: def __init__(self, config): self.db_host = config['host'] self.db_user = config['user'] self.db_password = config['password'] self.db_name = config['database'] self.backup_folder = config['backup_folder'] self.max_backups = config.get('max_backups', 30) # 默认保留30个备份 # 验证备份目录 Path(self.backup_folder).mkdir(parents=True, exist_ok=True) def _get_timestamp(self): """生成标准时知携誉间戳""" return datetime.now().strftime('%Y-%m-%d_%H-%M-%S') def _clean_old_backups(self): """清理旧备份文件""" try: backups = sorted(Path(self.backup_folder).glob(f"{self.db_name}_*.sql")) if len(backups) >= self.max_backups: for old_file in backups[:-(self.max_backups - 1)]: old_file.unlink() logging.info(f"Deleted old backup: {old_file.name}") except Exception as e: logging.error(f"Error cleaning old backups: {str(e)}") def backup(self): """执行数据库备份""" try: # 测试数据库搭段连隐烂接 with pymysql.connect( host=self.db_host, user=self.db_user, password=self.db_password, database=self.db_name ) as conn: conn.ping(reconnect=True) timestamp = self._get_timestamp() backup_file = Path(self.backup_folder) / f"{self.db_name}_{timestamp}.sql" # 使用mysqldump命令备份 cmd = ( f"mysqldump -h {self.db_host} -u {self.db_user} " f"-p{self.db_password} --single-transaction " f"--routines --triggers {self.db_name} > {backup_file}" ) exit_code = os.system(cmd) if exit_code == 0: logging.info(f"Backup successful: {backup_file.name}") self._clean_old_backups() return True else: raise RuntimeError(f"mysqldump failed with exit code {exit_code}") except pymysql.Error as e: logging.error(f"Database connection error: {str(e)}") except Exception as e: logging.error(f"Backup failed: {str(e)}") return False# 使用示例if __name__ == "__main__": config = { 'host': 'localhost', 'user': 'username', 'password': 'password', 'database': 'database_name', 'backup_folder': '/path/to/backups', 'max_backups': 14 # 保留14天备份 } backup = MySQLBackup(config) backup.backup()关键改进说明错误处理增强
添加数据库连接测试
捕获mysqldump执行结果（通过退出码）
完善的日志记录系统
备份文件管理
自动清理旧备份（可配置保留数量）
使用Path对象处理路径（跨平台兼容）
确保备份目录存在
备份质量提升
添加--single-transaction选项保证一致性（InnoDB）
包含存储过程和触发器（--routines --triggers）
标准化的时间戳命名
安全注意事项
建议将密码存储在环境变量或配置文件中（不要硬编码）
备份文件权限应限制为仅管理员可读
- 定时任务设置（Linux crontab示例）：
- Windows任务计划程序：
- 创建基本任务
- 设置触发器为"每日"
- 操作为"启动程序"，指向python.exe和脚本路径
- 监控建议
- 设置日志监控（如通过logrotate管理日志文件）
- 添加邮件/短信报警（当连续多次备份失败时）
建议定期测试备份文件的可用性：
def test_restore(backup_file, test_db_name): """测试备份文件是否可恢复""" try: # 创建临时测试数据库 with pymysql.connect(host='localhost', user='root') as conn: conn.cursor().execute(f"CREATE DATABASE IF NOT EXISTS {test_db_name}") # 执行恢复 cmd = f"mysql -h localhost -u root {test_db_name} < {backup_file}" if os.system(cmd) == 0: print("Restore test successful") return True except Exception as e: print(f"Restore test failed: {str(e)}") return False此方案提供了企业级数据库备份的基本框架，可根据实际需求进一步扩展（如添加压缩、加密、远程上传等功能）。
热门标签
