使用Python自制數(shù)據(jù)庫備份工具實現(xiàn)數(shù)據(jù)定時覆蓋

更新時間：2024年03月26日 08:36:52 作者：IT小輝同學

這篇文章主要為大家詳細介紹了如何使用Python自制數(shù)據(jù)庫備份工具實現(xiàn)數(shù)據(jù)定時覆蓋功能,文中的示例代碼講解詳細,需要的小伙伴可以參考下

工具介紹

自動化測試數(shù)據(jù)庫更新調(diào)度程序

這段 Python 腳本自動化了每天定時從生產(chǎn)數(shù)據(jù)庫更新測試數(shù)據(jù)庫的過程。它利用了 schedule 庫來安排并執(zhí)行每天指定時間的更新任務。

特點

自動數(shù)據(jù)庫更新: 腳本自動連接到生產(chǎn)數(shù)據(jù)庫，檢索所有表，并將它們的數(shù)據(jù)轉(zhuǎn)移到測試數(shù)據(jù)庫。
日志記錄: 實現(xiàn)了全面的日志記錄，以跟蹤執(zhí)行狀態(tài)和更新過程中可能發(fā)生的任何錯誤。日志存儲在 update_test_db.log 文件中。
批處理處理: 數(shù)據(jù)傳輸分批進行，以優(yōu)化性能，確保對大型數(shù)據(jù)集的高效處理。

如何使用

配置:

在 prod_db_config 和 test_db_config 字典中配置生產(chǎn)和測試數(shù)據(jù)庫的連接參數(shù)，包括用戶名、密碼、IP 地址、數(shù)據(jù)庫名稱和端口。

依賴項:

確保已安裝所需的依賴項。如果沒有，請使用 pip install -r requirements.txt 進行安裝。
cd ./static
pip install .\mysqlclient-1.4.6-cp37-cp37m-win_amd64.whl

日志記錄:

日志寫入到 update_test_db.log 文件中。確保腳本對其所在目錄具有寫權限。

調(diào)度:

更新任務被安排在每天午夜（00:00）執(zhí)行。您可以通過修改 schedule.every().day.at("00:00").do(update_test_db) 行來調(diào)整計劃。

執(zhí)行:

運行腳本。它將每 60 秒檢查一次是否有待處理任務。

監(jiān)控:

監(jiān)視日志文件 (update_test_db.log)，以跟蹤執(zhí)行狀態(tài)和更新過程中可能遇到的任何潛在錯誤。

注意

數(shù)據(jù)完整性:

確保生產(chǎn)數(shù)據(jù)庫包含必要的數(shù)據(jù)，并且測試數(shù)據(jù)庫僅用于測試目的，以避免意外修改。

安全性:

保護數(shù)據(jù)庫憑據(jù)，并限制對敏感信息的訪問。

錯誤處理:

腳本包含了健壯的錯誤處理機制，以捕獲并記錄執(zhí)行過程中可能出現(xiàn)的任何異常。

打包應用程序

您可以使用 PyInstaller 將腳本打包成一個可執(zhí)行文件，并且可以指定一個圖標作為應用程序的圖標。例如，您可以運行以下命令：

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

這將創(chuàng)建一個獨立的可執(zhí)行文件，用戶可以直接運行而無需安裝 Python 或其他依賴項。

貢獻:

歡迎對腳本進行貢獻和改進。請隨時 fork 倉庫，進行修改，并提交 pull 請求。

關于:

該腳本旨在簡化測試數(shù)據(jù)庫更新的過程，促進更順暢的開發(fā)和測試工作流程。

核心源碼

import schedule
import time
import pandas as pd
from sqlalchemy import create_engine, text

# 生產(chǎn)數(shù)據(jù)庫配置
prod_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test1',
    'port': 3306
}

# 測試數(shù)據(jù)庫配置
test_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test2',
    'port': 3306
}


def update_test_db():
    try:
        # 構建測試數(shù)據(jù)庫連接字符串
        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"

        # 創(chuàng)建測試數(shù)據(jù)庫引擎
        test_engine = create_engine(test_db_url)

        # 測試連接是否成功
        with test_engine.connect() as conn:
            conn.execute(text("select 1"))
        print("成功連接到測試數(shù)據(jù)庫.")

        # 清除測試數(shù)據(jù)庫中的所有表
        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]}")

        # 獲取生產(chǎn)數(shù)據(jù)庫中的所有表
        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("從生產(chǎn)數(shù)據(jù)庫中獲取表完成.")

            # 將生產(chǎn)數(shù)據(jù)庫中的所有表數(shù)據(jù)備份到測試數(shù)據(jù)庫
            for table in tables:
                table_name = table[0]
                print(f"備份表 {table_name} 數(shù)據(jù)...")
                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} 數(shù)據(jù)備份完成.")

        print("測試數(shù)據(jù)庫更新成功.")
    except Exception as e:
        print("更新測試數(shù)據(jù)庫時出錯:", e)

if __name__ == "__main__":
    # 將任務調(diào)度在每天特定時間執(zhí)行
    schedule.every().day.at("23:22").do(update_test_db)

    # 運行定時任務
    while True:
        schedule.run_pending()
        time.sleep(60)