数据库备份东西（完结数据守时掩盖）

数据库备份东西（完结数据守时掩盖）

永久热爱，永久执着！

东西介绍

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

这段 Python 脚本自动化了每天守时从出产数据库更新测验数据库的进程。它利用了 schedule 库来组织并履行每天指守时刻的更新使命。

特色:

  • 自动数据库更新: 脚本自动衔接到出产数据库，检索一切表，并将它们的数据转移到测验数据库。

  • 日志记载: 完结了全面的日志记载，以盯梢履行状况和更新进程中可能发生的任何过错。日志存储在 update_test_db.log 文件中。

  • 批处理处理: 数据传输分批进行，以优化功能，保证对大型数据集的高效处理。

怎么运用:

  1. 装备:

    • 在 prod_db_config 和 test_db_config 字典中装备出产和测验数据库的衔接参数，包括用户名、暗码、IP 地址、数据库名称和端口。

  2. 依靠项:

    • 保证已装置所需的依靠项。如果没有，请运用 pip install -r requirements.txt 进行装置。
    • cd ./static
    • pip install .mysqlclient-1.4.6-cp37-cp37m-win_amd64.whl

  3. 日志记载:

    • 日志写入到 update_test_db.log 文件中。保证脚本对其地点目录具有写权限。

  4. 调度:

    • 更新使命被组织在每天午夜（00:00）履行。您能够通过修正 schedule.every().day.at("00:00").do(update_test_db) 行来调整方案。

  5. 履行:

    • 运转脚本。它将每 60 秒检查一次是否有待处理使命。

  6. 监控:

    • 监视日志文件 (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)

运转作用

1.可直接代码运转

数据库备份东西（完结数据守时掩盖）

1.可履行exe程序

数据库备份东西（完结数据守时掩盖）