Python基于docker部署的Mysql備份查詢腳本

更新時間：2024年04月19日 10:01:25 作者：蝸牛其實也很努力

這篇文章主要來和大家分享Python基于docker部署的Mysql備份查詢的腳本,文中的示例代碼講解詳細,有需要的小伙伴可以跟隨小編一起了解下

前言

此環(huán)境是基于docker部署的mysql，docker部署mysql可以參考如下鏈接：

docker 部署服務案例

顏色塊文件

root@bogon:~ 2024-04-18 16:34:23# cat DefaultColor.py
#########################################################################
#    File Name: DefaultColor.py
#    Author: eight
#    Mail: 18847097110@163.com 
#    Created Time: Thu 11 Apr 2024 10:25:31 PM CST
#########################################################################
#!/usr/bin/env python
# -*- coding: utf-8 -*-
class Color:
        END = '\033[0m'  # normal
        BOLD = '\033[1m'  # bold
        RED = '\033[1;91m'  # red
        GREEN = '\033[1;92m'  # green
        ORANGE = '\033[1;93m'  # orange
        BLUE = '\033[1;94m'  # blue
        PURPLE = '\033[1;95m'  # purple
        UNDERLINE = '\033[4m'  # underline
        CYAN = '\033[1;96m'  # cyan
        GREY = '\033[1;97m'  # gray
        BR = '\033[1;97;41m'  # background red
        BG = '\033[1;97;42m'  # background green
        BY = '\033[1;97;43m'  # background yellow

備份代碼

 
#########################################################################
#    File Name: query_mysql.py
#    Author: eight
#    Mail: 18847097110@163.com 
#    Created Time: Thu 18 Apr 2024 11:12:26 AM CST
#########################################################################
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
import configparser
import os
from datetime import datetime
import DefaultColor
import subprocess
 
 
# 創(chuàng)建備份目錄
def mkdir():
        dir_path = os.getcwd()
        print("--------------------------------------------------------")
        print("當前腳本工作目錄：", dir_path)
        mysql_backup_dir = "/opt/mysql_backup"
        print("--------------------------------------------------------")
        if not os.path.exists(mysql_backup_dir):
                os.makedirs(mysql_backup_dir)
        else:
                return f"數(shù)據(jù)庫備份目錄 '{mysql_backup_dir}' 已經(jīng)存在，跳過創(chuàng)建"
        return "數(shù)據(jù)庫備份目錄：" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 創(chuàng)建成功"
 
 
 
# 備份數(shù)據(jù)庫 
def backup(mysql_backup_dir):
        config = configparser.ConfigParser()
        config.read("config.ini")
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        current_datetime = datetime.now()
        formatted_datetime = current_datetime.strftime("%Y-%m-%d_%H-%M-%S")
        try:
                command = f'docker exec -it mysql-container /bin/bash -c "mysqldump -h {host} -u {username} -p{password} -P {port} -- {db} > {mysql_backup_dir}-{db}-{formatted_datetime}.sql"'
                subprocess.run(command, shell=True, check=True)
                
        except subprocess.CalledProcessError as e:
                print("Database backup failed:", e)
        return f"數(shù)據(jù)庫：{db}備份完成: {mysql_backup_dir}-{db}-{formatted_datetime}.sql"
 
 
 
if __name__ == '__main__':
        backup_dir = mkdir()
        print(backup_dir)
        backup = backup("/opt/mysql_backup")
        print(backup)

此代碼如果是本地環(huán)境測試的話，邏輯是判斷當前本地有沒有存儲備份數(shù)據(jù)的目錄，沒有則創(chuàng)建。

再利用subprocess執(zhí)行mysqldump命令，按照日期生成.sql文件

效果

因為是docker環(huán)境，所以判斷當前目錄是否存在，是判斷的宿主機上的，所以此代碼不是很嚴謹?shù)拇a，如果是本地部署的話，把mysqldump命令前面的 docker exec -it mysql-container /bin/bash -c 刪除掉即可。

這樣備份的腳本就實現(xiàn)了腳本預期效果：將數(shù)據(jù)庫備份到指定的目錄下，以日期區(qū)別sql文件

全量代碼（讀取配置和備份和查詢）

root@bogon:~ 2024-04-18 16:30:14# cat MysqlBackup.py 
#########################################################################
#    File Name: query_mysql.py
#    Author: eight
#    Mail: 18847097110@163.com 
#    Created Time: Thu 18 Apr 2024 11:12:26 AM CST
#########################################################################
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
import configparser
import os
from datetime import datetime
import DefaultColor
import subprocess
 
# 讀取mysql配置文件
def mysql_config():
        config = configparser.ConfigParser()
        config.read("config.ini")
 
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        print("--------------------------------------------------------")
        return f"主機：{host}\t用戶名:{username}\t密碼:{password}\t端口:{port}\t數(shù)據(jù)庫:{db}"
 
 
# 創(chuàng)建備份目錄
def mkdir():
        dir_path = os.getcwd()
        print("--------------------------------------------------------")
        print("當前腳本工作目錄：", dir_path)
        mysql_backup_dir = "/opt/mysql_backup"
        print("--------------------------------------------------------")
        if not os.path.exists(mysql_backup_dir):
                os.makedirs(mysql_backup_dir)
        else:
                return f"數(shù)據(jù)庫備份目錄 '{mysql_backup_dir}' 已經(jīng)存在，跳過創(chuàng)建"
        return "數(shù)據(jù)庫備份目錄：" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 創(chuàng)建成功"
 
 
# 查詢sql語句
##def select():
#       config = configparser.ConfigParser()
#       config.read("config.ini")
#                    
#       # 獲取配置信息
#       host = config.get("MYSQL", "MYSQL_HOST")
#       username = config.get("MYSQL", "MYSQL_USER")
#       password = config.get("MYSQL", "MYSQL_PASSWORD")
#       port = config.get("MYSQL", "MYSQL_PORT")
#       db = config.get("MYSQL", "MYSQL_DB")
#       # 創(chuàng)建mysql連接信息
#       conn = pymysql.connect(host=host, user=username, password=password, port=int(port), db=db)
#       cur = conn.cursor()
#       # 執(zhí)行查詢語句
#       cur.execute("use docker;")
#       cur.execute("select * from DockerImages;")
#       print("查詢到的數(shù)據(jù)是：")
#       # 接收全部的返回結果行
#       result = cur.fetchall()
#       while True:
#               resp = cur.fetchone()
#               if resp is None:
#                       break
#       # 關閉查詢游標
#       cur.close()
#       # 提交
#       conn.commit()
#       # 關閉鏈接，釋放計算機資源
#       conn.close()
#       return result
 
 
# 備份數(shù)據(jù)庫 
def backup(mysql_backup_dir):
        config = configparser.ConfigParser()
        config.read("config.ini")
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        current_datetime = datetime.now()
        formatted_datetime = current_datetime.strftime("%Y-%m-%d_%H-%M-%S")
        try:
                command = f'docker exec -it mysql-container /bin/bash -c "mysqldump -h {host} -u {username} -p{password} -P {port} -- {db} > {mysql_backup_dir}-{db}-{formatted_datetime}.sql"'
                subprocess.run(command, shell=True, check=True)
                
        except subprocess.CalledProcessError as e:
                print("Database backup failed:", e)
        return f"數(shù)據(jù)庫：{db}備份完成: {mysql_backup_dir}-{db}-{formatted_datetime}.sql"
 
 
 
if __name__ == '__main__':
        #mysql_config = mysql_config()
        #print(mysql_config)
        backup_dir = mkdir()
        print(backup_dir)
 
        #sector = select()
        # 循環(huán)打印所有數(shù)據(jù)，如果此代碼是封裝在函數(shù)內部，那么使用return的話，只會輸出第一條數(shù)據(jù)，所以要在外部調用
        #for row in sector:
        #       print(row)
        backup = backup("/opt/mysql_backup")
        print(backup)
root@bogon:~ 2024-04-18 16:30:18# 
root@bogon:~ 2024-04-18 16:30:32# 
root@bogon:~ 2024-04-18 16:30:32# cat MysqlBackup.py 
#########################################################################
#    File Name: query_mysql.py
#    Author: eight
#    Mail: 18847097110@163.com 
#    Created Time: Thu 18 Apr 2024 11:12:26 AM CST
#########################################################################
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
import configparser
import os
from datetime import datetime
import DefaultColor
import subprocess
 
# 讀取mysql配置文件
def mysql_config():
        config = configparser.ConfigParser()
        config.read("config.ini")
 
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        print("--------------------------------------------------------")
        return f"主機：{host}\t用戶名:{username}\t密碼:{password}\t端口:{port}\t數(shù)據(jù)庫:{db}"
 
 
# 創(chuàng)建備份目錄
def mkdir():
        dir_path = os.getcwd()
        print("--------------------------------------------------------")
        print("當前腳本工作目錄：", dir_path)
        mysql_backup_dir = "/opt/mysql_backup"
        print("--------------------------------------------------------")
        if not os.path.exists(mysql_backup_dir):
                os.makedirs(mysql_backup_dir)
        else:
                return f"數(shù)據(jù)庫備份目錄 '{mysql_backup_dir}' 已經(jīng)存在，跳過創(chuàng)建"
        return "數(shù)據(jù)庫備份目錄：" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 創(chuàng)建成功"
 
 
# 查詢sql語句
def select():
        config = configparser.ConfigParser()
        config.read("config.ini")
                     
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        # 創(chuàng)建mysql連接信息
        conn = pymysql.connect(host=host, user=username, password=password, port=int(port), db=db)
        cur = conn.cursor()
        # 執(zhí)行查詢語句
        cur.execute("use docker;")
        cur.execute("select * from DockerImages;")
        print("查詢到的數(shù)據(jù)是：")
        # 接收全部的返回結果行
        result = cur.fetchall()
        while True:
                resp = cur.fetchone()
                if resp is None:
                        break
        # 關閉查詢游標
        cur.close()
        # 提交
        conn.commit()
        # 關閉鏈接，釋放計算機資源
        conn.close()
        return result
 
 
# 備份數(shù)據(jù)庫 
def backup(mysql_backup_dir):
        config = configparser.ConfigParser()
        config.read("config.ini")
        # 獲取配置信息
        host = config.get("MYSQL", "MYSQL_HOST")
        username = config.get("MYSQL", "MYSQL_USER")
        password = config.get("MYSQL", "MYSQL_PASSWORD")
        port = config.get("MYSQL", "MYSQL_PORT")
        db = config.get("MYSQL", "MYSQL_DB")
        current_datetime = datetime.now()
        formatted_datetime = current_datetime.strftime("%Y-%m-%d_%H-%M-%S")
        try:
                command = f'docker exec -it mysql-container /bin/bash -c "mysqldump -h {host} -u {username} -p{password} -P {port} -- {db} > {mysql_backup_dir}-{db}-{formatted_datetime}.sql"'
                subprocess.run(command, shell=True, check=True)
                
        except subprocess.CalledProcessError as e:
                print("Database backup failed:", e)
        return f"數(shù)據(jù)庫：{db}備份完成: {mysql_backup_dir}-{db}-{formatted_datetime}.sql"
 
 
 
if __name__ == '__main__':
        mysql_config = mysql_config()
        print(mysql_config)
        backup_dir = mkdir()
        print(backup_dir)
 
        sector = select()
        #循環(huán)打印所有數(shù)據(jù)，如果此代碼是封裝在函數(shù)內部，那么使用return的話，只會輸出第一條數(shù)據(jù)，所以要在外部調用
        for row in sector:
                print(row)
        backup = backup("/opt/mysql_backup")
        print(backup)