前言
此环境是基于docker部署的mysql,docker部署mysql可以参考如下链接:
docker 部署服务案例-CSDN博客
颜色块文件
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 = '3[0m' # normal BOLD = '3[1m' # bold RED = '3[1;91m' # red GREEN = '3[1;92m' # green ORANGE = '3[1;93m' # orange BLUE = '3[1;94m' # blue PURPLE = '3[1;95m' # purple UNDERLINE = '3[4m' # underline CYAN = '3[1;96m' # cyan GREY = '3[1;97m' # gray BR = '3[1;97;41m' # background red BG = '3[1;97;42m' # background green BY = '3[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 # 创建备份目录 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"数据库备份目录 '{mysql_backup_dir}' 已经存在,跳过创建" return "数据库备份目录:" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 创建成功" # 备份数据库 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"数据库:{db}备份完成: {mysql_backup_dir}-{db}-{formatted_datetime}.sql" if __name__ == '__main__': backup_dir = mkdir() print(backup_dir) backup = backup("/opt/mysql_backup") print(backup)
此代码如果是本地环境测试的话,逻辑是判断当前本地有没有存储备份数据的目录,没有则创建。
再利用subprocess执行mysqldump命令,按照日期生成.sql文件
效果
因为是docker环境,所以判断当前目录是否存在,是判断的宿主机上的,所以此代码不是很严谨的代码,如果是本地部署的话,把mysqldump命令前面的 docker exec -it mysql-container /bin/bash -c 删除掉即可。
这样备份的脚本就实现了脚本预期效果:将数据库备份到指定的目录下,以日期区别sql文件
全量代码(读取配置和备份和查询)
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数据库:{db}" # 创建备份目录 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"数据库备份目录 '{mysql_backup_dir}' 已经存在,跳过创建" return "数据库备份目录:" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 创建成功" # 查询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") # 创建mysql连接信息 conn = pymysql.connect(host=host, user=username, password=password, port=int(port), db=db) cur = conn.cursor() # 执行查询语句 cur.execute("use docker;") cur.execute("select * from DockerImages;") print("查询到的数据是:") # 接收全部的返回结果行 result = cur.fetchall() while True: resp = cur.fetchone() if resp is None: break # 关闭查询游标 cur.close() # 提交 conn.commit() # 关闭链接,释放计算机资源 conn.close() return result # 备份数据库 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"数据库:{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() #循环打印所有数据,如果此代码是封装在函数内部,那么使用return的话,只会输出第一条数据,所以要在外部调用 for row in sector: print(row) backup = backup("/opt/mysql_backup") print(backup)
效果
优化
将配置文件信息以字典的方式返回
在查询函数时,使用相应的键来访问这些值
避免重复定义变量
######################################################################### # 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 { "host": host, "username": username, "password": password, "port": port, "db": db } # 创建备份目录 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"数据库备份目录 '{mysql_backup_dir}' 已经存在,跳过创建" return "数据库备份目录:" + DefaultColor.Color.GREEN + mysql_backup_dir + DefaultColor.Color.END + " 创建成功" # 查询sql语句 def select(): mysql_connect_info = mysql_config() # 创建mysql连接信息 conn = pymysql.connect(host=mysql_connect_info["host"], user=mysql_connect_info["username"], password=mysql_connect_info["password"], port=int(mysql_connect_info["port"]), db=mysql_connect_info["db"]) cur = conn.cursor() # 执行查询语句 cur.execute("use docker;") cur.execute("select * from DockerImages;") print("查询到的数据是:") # 接收全部的返回结果行 result = cur.fetchall() while True: resp = cur.fetchone() if resp is None: break # 关闭查询游标 cur.close() # 提交 conn.commit() # 关闭链接,释放计算机资源 conn.close() return result # 备份数据库 def backup(mysql_backup_dir): # 获取配置信息 current_datetime = datetime.now() formatted_datetime = current_datetime.strftime("%Y-%m-%d_%H-%M-%S") mysql_connect_info = mysql_config() try: command = f'docker exec -it mysql-container /bin/bash -c "mysqldump -h {mysql_connect_info["host"]} -u {mysql_connect_info["username"]} -p{mysql_connect_info["password"]} -P {mysql_connect_info["port"]} -- {mysql_connect_info["db"]} > {mysql_backup_dir}-{mysql_connect_info["db"]}-{formatted_datetime}.sql"' subprocess.run(command, shell=True, check=True) except subprocess.CalledProcessError as e: print("Database backup failed:", e) return f'数据库:{mysql_connect_info["db"]}备份完成: {mysql_backup_dir}-{mysql_connect_info["db"]}-{formatted_datetime}.sql' if __name__ == '__main__': backup_dir = mkdir() print(backup_dir) sector = select() #循环打印所有数据,如果此代码是封装在函数内部,那么使用return的话,只会输出第一条数据,所以要在外部调用 for row in sector: print(row) backup = backup("/opt/mysql_backup") print(backup)
文章版权声明:除非注明,否则均为VPS857原创文章,转载或复制请以超链接形式并注明出处。
还没有评论,来说两句吧...