Python实现MySQL DBA小工具一例
#!/bin/env python3"""
by lijiankai 20160602
"""
from sys import exit
from time import sleep
from argparse import ArgumentParser,RawTextHelpFormatter
import pymysql
#----------
#定义每种显示类型(由-t指定)所包含的状态
#添加和修改要展示的信息修改types_dic字典即可
#同时关注下面add_argument方法添加-t选项时其中choices参数的值，应该和types_dic的键一致
#----------
types_dic = {}
types_dic['qps'] = ('Com_select','Com_update','Com_insert','Com_delete','Com_replace','Questions','Queries')
types_dic['table_file'] = ('Opened_tables','Opened_table_definitions','Opened_files','Created_tmp_tables','Created_tmp_disk_tables','Created_tmp_files',)
types_dic['general'] = ('Slow_queries','Select_full_join','Select_full_range_join','Select_range','Select_scan','Sort_range','Sort_rows','Sort_scan')
types_dic['threading'] = ('threads_created')
types_dic['networking_stats'] = ('Bytes_received','Bytes_sent','Connections','Aborted_connects')
types_dic['innodb_buffer_pool'] = ('Innodb_buffer_pool_read_requests','Innodb_buffer_pool_reads','Innodb_buffer_pool_write_requests','Innodb_buffer_pool_wait_free')
types_dic['innodb_data'] = ('Innodb_data_fsyncs','Innodb_data_read','Innodb_data_written')
types_dic['innodb_stats'] = ('Innodb_log_write_requests','Innodb_log_writes','Innodb_os_log_fsyncs','Innodb_os_log_writes')
#----------
#get_args()函数通过argparse模块的ArgumentParser类来生成帮助信息并获取命令行参数
#生成一个全局变量字典对象args，保存处理过的命令行参数
#----------
def get_args():
#实例化类，formatter_class参数允许help信息以自定义的格式显示
parser = ArgumentParser(description="This is a simple tool for MySQL DBA.\nWith this tool you can see several kinds of status's average or increase value in the last N seconds",formatter_class =RawTextHelpFormatter)
#group_necessary = parser.add_argument_group('necessary arguments')
#group_optional = parser.add_argument_group('optional arguments') #默认既有该项，所有参数均位于该项下
parser.add_argument('-u',metavar='USER',dest='user',help="mysql user",required=True)
parser.add_argument('-p',metavar='PASSWORD',dest='password',help="mysql password",required=True)
parser.add_argument('-H',metavar='HOSTNAME',dest='host',help="mysql hostname",required=True,)
parser.add_argument('-P',metavar='PORT',dest='port',help="mysql port(default 3306)",default=3306,type=int)
parser.add_argument('-i',metavar='INTERVAL_TIME',dest='interval',help="interval time(unit=second,default 10s)",default=10,type=int)
#--average表示是否查看平均值，默认显示差值 action不能和metavar共存
parser.add_argument('--average',dest='average',help='show average value in the interval time(default is the increase value)',action='store_true') #store_true不能喝metavar共存
#下面-t选项中的choices的列表值需和脚本开始处定义的types_dic字典的键一致
parser.add_argument('-t',metavar='INFORMATION_TYPE',dest='type',choices=['qps','table_file','general','threading','networking_stats','innodb_buffer_pool','innodb_data','innodb_stats'],help="""information type.See the allowed types below:
qps : numbers of the DML command executed
table_file : tables or tmp_tables or tmp_files that are opend or created
general : something about select
networking_stats : something about network or connect
innodb_buffer_pool : something about innodb_buffer_pool stats
innodb_data : something about innodb_data stats
innodb_stats : something about innodb stats""",\
required=True)
#全局字典 键(add_argument()中的dest):值(用户输入)
#vars将Namespace object转换成dict object
global args
args = vars(parser.parse_args())
#print(args)
#----------
#process_query()函数从get_args()返回值中拿到登陆mysql需要的相关信息
#执行show global status语句，并将结果保存在status_dict字典中
#----------
def process_query():
status_dict={} #存放所有status值
try:
with pymysql.connect(host=args['host'],user=args['user'],password=args['password'],charset='utf8',port=args['port']) as mysql_cur:
mysql_cur.execute('show global status') #执行语句，查询结果的每一行作为一个元组存进mysql_cur中
except pymysql.err.MySQLError as err:
print("ERROR: "+str(err))
exit(10)
for status in mysql_cur:
status_dict]=status #更新status_dict字典
return status_dict
#----------
#show_result()函数处理相关数据，展示最终结果
#----------
def show_result(type):
#打印头部
print()
for status in types_dic]:
print(' {}'.format(status),end='')
print()
#开始循环显示
try:
while True:
status_dic1=process_query()
sleep(args['interval'])
status_dic2=process_query()
#打印各值
for k in types_dic]:
if args['average'] is True:
#有--average选项，输出指定时间内的平均值
print( ' ' + str(round((int(status_dic2)-int(status_dic1))/args['interval'],2)).center(len(k)),end='' )
else:
#输出指定时间内的增长值
print( ' ' + str(round(int(status_dic2)-int(status_dic1))).center(len(k)),end='' )
print()
except KeyboardInterrupt:
print('\n-----bye-----')
if __name__ == '__main__':
get_args()
process_query()
show_result(args['type'])
页:
[1]