DbfMgr.py

#!/usr/bin/python3 -O
"""
 File: 				DbfMgr.py
 Python Version:		3.12.11
 Author: 			Leon McClatchey
 Company:			Linktech Engineering LLC
 Date: 				2025-10-26

 Description: 	Utilizes YAML files and Ansible Vaults to manage Database servers
				The parameter for the Host can be used to define the distro (group) as well
				But, when the Host is used to define the distro/group, 
				that distro must exist in the yaml data file (currently opensuse, ubuntu, and redhat)
 """
# System Libraries
import argparse
import os, sys
import logging
import pathlib
from datetime import datetime
from ansible_vault import Vault
# Initialize Global Strings
Config = {}
# Function to Backup Database(s)
def BackupDistros():
	global Config
	log = Config['Args']['log']['path']
	engine = Config['Args']['engine']
	age = Config['Hosts']['age']
	distro = Config['Args']['distro']
	psw = Config['Args']['password']
	servers = Config['Hosts'][engine]['servers'][distro]
	vault = AnsibleTools.AnsibleCredentials(Config['Args']['inventory'],psw,engine)
	user = vault['backups']['user']
	pswd = vault['backups']['pass']
	for host in servers.keys():
		qry = "show databases where `DATABASE` not in ('information_schema','performance_schema','mysql','sys');"
		args = {'host':host,'user':user,'password':pswd,'log':log,'query':qry,'engine':engine}
		if DbTools.TestConnection(**args) is True:
			pth = servers[host]['path'] if NetTools.isLocalHost(host) is True else Config['Defaults']['bkup-fldr']
			pth = os.path.join(pth,host,engine)
			rc = DbTools.GetList(**args)
			for dbf in rc:
				FileTools.CreateFolder(pth)
				ts = datetime.now().strftime("%Y%m%d")
				target = os.path.join(pth,f"{dbf['Database']}_{ts}.sql")
				FileTools.WriteLog(f"Dumping {dbf['Database']} to {target}",log)
				DbTools.DumpDatabase(host,user,pswd,dbf['Database'],target)
			if len(rc) > 0:
				FileTools.CompressFiles(pth,"*.sql",engine,log)
				FileTools.RemoveOldFiles(pth,age,log)
# Function to check the system Path
def CheckSysPath(target):
	npath = os.path.normcase(os.path.abspath(target))
	for p in sys.path:
	  nsys = os.path.normcase(os.path.abspath(p))
	  if npath == nsys: return True
	return False
# Function to configure the hosts
def ConfigureHosts():
	global Config
	yml = os.path.join(Config['Args']['inventory'],"hosts.yml")
	log = Config['Args']['log']['path']
	Ydata = AnsibleTools.LoadYaml(yml,log)
	engine = Config['Args']['engine']
	distro = [ Config['Args']['distro'] ]
	dbservers = Ydata.get('all',{}).get('children',{}).get('dbservers',{})
	kargs = {'data':Ydata,'log':log,'dbservers':dbservers,'distros':distro,'engine':engine,'host':Config['Args']['host']}
	AnsibleTools.InjectYamlPaths(**kargs)
	AnsibleTools.RepopulateDbfHosts(**kargs)
	Config['Hosts'] = Ydata['all']['children']['dbservers']
	Config['Defaults'] = {'bkup-fldr':Ydata['all']['bkup-fldr']['path']}
# Function to get the Commandline arguments
def GetArguments():
	global Config
	script_path = os.path.abspath(sys.argv[0])
	script_dir = os.path.dirname(script_path)
	script_name = os.path.basename(script_path)
	Config["Script"] = {"path":script_path,"folder":script_dir,"name":script_name}
	MyScript = f"{os.path.splitext(script_name)[0]}"
	parser = MyParser(
		prog=f"{MyScript}",
		description='Backup and Restores Databases',
		epilog='Either a Hostname or a Distro must be specified',
		script=f"{MyScript}",
		script_version='1.0.0',
		include_python_version=True,
		ansible_cfg=True,
		usage = "%(prog)s [-e -f -v -H -P -l -L -u -k -a -s -T] \
			\n(-h displays help, --version displays version)",
		formatter_class=argparse.RawTextHelpFormatter
	)
	parser.add_argument('-a','--action',dest='action',required=True,type=str,help='Operation being performed:\n "backup", "restore", or "list" or "view"')
	parser.add_argument('-e','--engine',dest='engine',required=True,type=str,help='Database Engine being backed up (mariadb, mysql, postgresql, oracle)')
	try:
		known,unknown = parser.parse_known_args()
		if '-h' in unknown or '--help' in unknown:
			help_text = parser.get_help_text()
			parser.print_help()
			sys.exit(2)
		args = vars(known)
		msg = None
		if args['action'] is not None and args['action'].lower() not in ['backup','restore','list','view']:
			msg = f"Action must be backup,restore,list,or view\nNot {args['action']}!"
		if args['engine'] is not None and args['engine'].lower() not in ['mariadb','mysql','postgresql','oracle']:
			msg = f"Database Engine must be mariadb,mysql,posgresql,oracle, not {args['engine']}"
		args['engine'] = 'mariadb' if args['engine'].lower() == 'mysql' else args['engine']
		args['log']=Tools.String2Dictionary(args['log'],'=')
		args['password'] = AnsibleTools.read_vault_password(args['password'],args['log']['path']) if os.path.isfile(args['password']) else args['password']
		args['inventory'] = os.path.join(Config['Script']['folder'],'Data') if args['inventory'] is None else args['inventory']
		if args['distro'] is not None and len(args['distro']) > 0:
			Ydata = AnsibleTools.LoadYaml(os.path.join(args['inventory'],'hosts.yml'),args['log']['path'])
			distros = AnsibleTools.extract_distros(Ydata)
			if len(distros) > 0 and args['distro'] not in distros:
				msg = f"{args['distro']} is not a valid distro {distros}"
		if args['distro'] is None and NetTools.host_exists(args['host']) is False:
			msg = f"{args['host']} is not a valid host and No Distros has been declared!"
		elif args['host'] is not None:
			args['host'] = NetTools.LocalHostName() if args['host'].lower().strip() == 'localhost' else args['host']
		if msg is not None:
			FileTools.WriteLog(msg,args['log']['path']) if args['log']['path'] is not None else print(msg)
			raise argparse.ArgumentError(None,msg)
		return args
	except argparse.ArgumentError as e:
		parser.error(str(e))
# Function to Find the local Common folder and append it to the library path
def InitializeModules():
	ScriptPath = pathlib.Path(__file__).parent.resolve()
	LibPath = f"{ScriptPath}/Lib"
	if not CheckSysPath(LibPath): sys.path.append(LibPath)
# Main Function
def main():
	global Config
	Config['Args'] = GetArguments()
	FileTools.ArchiveLog(Config['Args']['log'])
	logging.basicConfig(filename=Config['Args']['log']['path'],filemode='a',level=logging.WARN)
	ConfigureHosts()
	BackupDistros()
# Preload with User Libraries
InitializeModules()
import FileTools,NetTools,Tools,DbTools,AnsibleTools
from MyParser import MyParser
# Execute Script
if __name__ =="__main__":
    main()