石家庄最好的网站建设公司,专业微信网站建设多少钱,dede企业模板/dedecms蓝色企模板/php网站源码,怎么找网站数据库备份工具#xff08;实现数据定时覆盖#xff09; 永远热爱#xff0c;永远执着#xff01; 工具介绍
自动化测试数据库更新调度程序
这段 Python 脚本自动化了每天定时从生产数据库更新测试数据库的过程。它利用了 schedule 库来安排并执行每天指定时间的更新任务…数据库备份工具实现数据定时覆盖 永远热爱永远执着 工具介绍
自动化测试数据库更新调度程序
这段 Python 脚本自动化了每天定时从生产数据库更新测试数据库的过程。它利用了 schedule 库来安排并执行每天指定时间的更新任务。
特点: 自动数据库更新: 脚本自动连接到生产数据库检索所有表并将它们的数据转移到测试数据库。 日志记录: 实现了全面的日志记录以跟踪执行状态和更新过程中可能发生的任何错误。日志存储在 update_test_db.log 文件中。 批处理处理: 数据传输分批进行以优化性能确保对大型数据集的高效处理。
如何使用: 配置: 在 prod_db_config 和 test_db_config 字典中配置生产和测试数据库的连接参数包括用户名、密码、IP 地址、数据库名称和端口。 依赖项: 确保已安装所需的依赖项。如果没有请使用 pip install -r requirements.txt 进行安装。cd ./staticpip 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 --iconmy_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 fmysql://{test_db_config[username]}:{test_db_config[password]}{test_db_config[ip]}:{test_db_config[port]}/{test_db_config[database]}?charsetutf8mb4# 创建测试数据库引擎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(fDROP TABLE IF EXISTS {table[0]})# 获取生产数据库中的所有表prod_db_url fmysql://{prod_db_config[username]}:{prod_db_config[password]}{prod_db_config[ip]}:{prod_db_config[port]}/{prod_db_config[database]}?charsetutf8mb4prod_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, indexFalse, if_existsreplace)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)运行效果
1.可直接代码运行 1.可执行exe程序
