当前位置：首页 > news >正文

数据库备份工具（实现数据定时覆盖）

news 2025/7/22 11:12:52

数据库备份工具（实现数据定时覆盖）

在这里插入图片描述

永远热爱，永远执着！

工具介绍

自动化测试数据库更新调度程序

这段 Python 脚本自动化了每天定时从生产数据库更新测试数据库的过程。它利用了 schedule 库来安排并执行每天指定时间的更新任务。

特点:

自动数据库更新: 脚本自动连接到生产数据库，检索所有表，并将它们的数据转移到测试数据库。
日志记录: 实现了全面的日志记录，以跟踪执行状态和更新过程中可能发生的任何错误。日志存储在 update_test_db.log 文件中。
批处理处理: 数据传输分批进行，以优化性能，确保对大型数据集的高效处理。

如何使用:

配置:
- 在 prod_db_config 和 test_db_config 字典中配置生产和测试数据库的连接参数，包括用户名、密码、IP 地址、数据库名称和端口。
依赖项:
- 确保已安装所需的依赖项。如果没有，请使用 pip install -r requirements.txt 进行安装。
- cd ./static
- pip install .\mysqlclient-1.4.6-cp37-cp37m-win_amd64.whl
日志记录:
- 日志写入到 update_test_db.log 文件中。确保脚本对其所在目录具有写权限。
调度:
- 更新任务被安排在每天午夜（00:00）执行。您可以通过修改 schedule.every().day.at("00:00").do(update_test_db) 行来调整计划。
执行:
- 运行脚本。它将每 60 秒检查一次是否有待处理任务。
监控:
- 监视日志文件 (update_test_db.log)，以跟踪执行状态和更新过程中可能遇到的任何潜在错误。

注意:

数据完整性:
- 确保生产数据库包含必要的数据，并且测试数据库仅用于测试目的，以避免意外修改。
安全性:
- 保护数据库凭据，并限制对敏感信息的访问。
错误处理:
- 脚本包含了健壮的错误处理机制，以捕获并记录执行过程中可能出现的任何异常。

打包应用程序:

您可以使用 PyInstaller 将脚本打包成一个可执行文件，并且可以指定一个图标作为应用程序的图标。例如，您可以运行以下命令：

pyinstaller --onefile --icon=my_icon.ico your_script.py

这将创建一个独立的可执行文件，用户可以直接运行而无需安装 Python 或其他依赖项。

贡献:

欢迎对脚本进行贡献和改进。请随时 fork 仓库，进行修改，并提交 pull 请求。

关于:

该脚本旨在简化测试数据库更新的过程，促进更顺畅的开发和测试工作流程。

祝愉快的测试! 🚀

核心源码

import schedule
import time
import pandas as pd
from sqlalchemy import create_engine, text# 生产数据库配置
prod_db_config = {'username': 'root','password': 'root','ip': '127.0.0.1','database': 'test1','port': 3306
}# 测试数据库配置
test_db_config = {'username': 'root','password': 'root','ip': '127.0.0.1','database': 'test2','port': 3306
}def update_test_db():try:# 构建测试数据库连接字符串test_db_url = f"mysql://{test_db_config['username']}:{test_db_config['password']}@{test_db_config['ip']}:{test_db_config['port']}/{test_db_config['database']}?charset=utf8mb4"# 创建测试数据库引擎test_engine = create_engine(test_db_url)# 测试连接是否成功with test_engine.connect() as conn:conn.execute(text("select 1"))print("成功连接到测试数据库.")# 清除测试数据库中的所有表with test_engine.connect() as conn:tables = conn.execute("SHOW TABLES").fetchall()if tables:for table in tables:conn.execute(f"DROP TABLE IF EXISTS {table[0]}")# 获取生产数据库中的所有表prod_db_url = f"mysql://{prod_db_config['username']}:{prod_db_config['password']}@{prod_db_config['ip']}:{prod_db_config['port']}/{prod_db_config['database']}?charset=utf8mb4"prod_engine = create_engine(prod_db_url)with prod_engine.connect() as prod_conn:tables = prod_conn.execute("SHOW TABLES").fetchall()print("从生产数据库中获取表完成.")# 将生产数据库中的所有表数据备份到测试数据库for table in tables:table_name = table[0]print(f"备份表 {table_name} 数据...")data = pd.read_sql_table(table_name, prod_conn)data.to_sql(table_name, test_engine, index=False, if_exists='replace')print(f"表 {table_name} 数据备份完成.")print("测试数据库更新成功.")except Exception as e:print("更新测试数据库时出错:", e)if __name__ == "__main__":# 将任务调度在每天特定时间执行schedule.every().day.at("23:22").do(update_test_db)# 运行定时任务while True:schedule.run_pending()time.sleep(60)