[python运维] 使用python3制作一个mysql压测小工具！
2024-06-04 02:36:15
0x01 argparse模块
argparse是Python 标准库中推荐的命令行解析模块，他可以实现给出参数提示和接收用户输入的参数，类似linux的命令行功能：
[root@yunweisn ~]# grepUsage: grep [OPTION]... PATTERN [FILE]...Try 'grep --help' for more information.[root@yunweisn ~]# grep --helpUsage: grep [OPTION]... PATTERN [FILE]...Search for PATTERN in each FILE or standard input.PATTERN is, by default, a basic regular expression (BRE).Example: grep -i 'hello world' menu.h main.c
Regexp selection and interpretation: -E, --extended-regexp PATTERN is an extended regular expression (ERE) -F, --fixed-strings PATTERN is a set of newline-separated fixed stringsargparse快速入门：
#!/usr/bin/python3import argparseparse = argparse.ArgumentParser(description='this is test script') # 实例化# description 描述信息parse.add_argument('-t','--test',dest='teststr',action='store', type=str,default='testing...',help='test help string') # 增加一个参数# -t 短参数# --test 长参数# dest 后续引用的变量名，不写默认是长参数名称# action有6种，不指定默认为store# type 默认为str，常见的还有int# default 参数的默认值# help 用户使用help命令时显示的信息args = parse.parse_args() # 接收用户输入的参数print(args.teststr)
# 运行1[root@yunweisn testdemo]# python3 testdemo01.py --helpusage: testdemo01.py [-h] [-t TESTSTR]
this is test script # 描述信息
optional arguments: -h, --help show this help message and exit # 默认有个help参数 -t TESTSTR, --test TESTSTR # 参数名称 test help string # 参数帮助
# 运行2[root@yunweisn testdemo]# python3 testdemo01.pytesting...[root@yunweisn testdemo]# python3 testdemo01.py -t 123123argparse有6种action
store 保存参数值，可能会先将参数值转换成另一个数据类型。若没有显式指定动作，则默认为该动作。
store_const 保存一个被定义为参数规格一部分的值，而不是一个来自参数解析而来的值。这通常用于实现非布尔值的命令行标记。
store_ture/store_false 保存相应的布尔值。这两个动作被用于实现布尔开关。
append 将值保存到一个列表中。若参数重复出现，则保存多个值。
append_const 将一个定义在参数规格中的值保存到一个列表中。
version 打印关于程序的版本信息，然后退出
0x02 来打造一个类mysql命令行客户端工具
先看看mysql命令默认是如何提示的？
[root@yunweisn ~]# mysqlERROR 1045 (28000): Access denied for user 'root'@'localhost' (using password: NO)
# 这里可见执行mysql的时候，有两个默认参数 # -u 用户名# -h 主机名python实现类似功能如下：
import argparseimport commands
def main(): parse = argparse.ArgumentParser(description='mysql client tools') parse.add_argument('-u','--user',dest='username',default='root',action='store_true',help='login to mysql username') parse.add_argument('-H','--host',dest='hostname',default='localhost',action='store_true',help='login to mysql server hostname') args = parse.parse_args() cmd = "mysql -u{} -p{}".format(args.username, args.hostname) print(commands.getoutput(cmd))
if __name__ == '__main__': main()# 运行01[root@yunweisn testdemo]# python3 mysql-client01.py ERROR 1045 (28000): Access denied for user 'root'@'localhost' (using password: YES)0x03 pymysql模块
pymysql用于在python中连接和操作mysql数据库。
快速入门：
#!/usr/bin/python3import pymysql
def main(): # 创建连接 conn = pymysql.connect(host='localhost,user='root',password='root',port=3306) # 创建一个游标 cursor = conn.cursor() # 执行sql语句 cursor.execute('show processlist;') # 获取所有的返回值 print(cursor.fetchall()) # 关闭 conn.close()
if __name__ == '__main__': main()0x04 faker模块伪造数据
faker可以生成一些看起来比较真实的数据
>>> from faker import Faker>>> faker = Faker("zh_CN")>>> faker.name()'莫玉珍'>>> faker.job()'人事经理'>>> faker.address()'甘肃省郑州市海港兰路E座 158840'0x05 简单实现一下插入数据
#！/usr/bin/python3#conding=utf-8import argparseimport pymysqlfrom faker import Faker import sys
def args_list(): parse = argparse.ArgumentParser('connect to mysql tools') parse.add_argument("--host",dest='hostname',action='store',default='localhost',help='mysql server hostname.') parse.add_argument("-u","--user",dest='username',action='store',default='root',help='connect to mysql server user.') parse.add_argument("-p","--password",dest='password',action='store',help='connect to mysql server password.') parse.add_argument("-P","--port",dest='port',action='store',default=3306,type=int,help='mysql server port.') parse.add_argument("-n","--linenumber",dest='linenumber',default=5,type=int,help='insert data line')
args = parse.parse_args() return args
def conn_db(**kwargs): try: # 连接数据库 conn = pymysql.connect(**kwargs) return conn # 返回conn except Exception as err: print("连接数据库错误,err=",err) sys.exit(1)def create_table(cursor): sql = """ create table if not exists `test`.`test001` (id int(10) not null auto_increment, name varchar(255) not null, job varchar(255) not null, address varchar(255) not null, primary key (`id`)) """ try: cursor.execute(sql) except Exception as err: print("创建表test001错误,err=",err) sys.exit(2)
def insert_data(cursor,linenumber,faker): for i in range(linenumber): # 根据用户输入的行数插入数据 name = faker.name() # 生成名字 job = faker.job() # 生成职业 address = faker.address() # 生成地址 # 拼接sql语句 sql = """ insert into `test`.`test001`(name,job,address) values('{0}','{1}','{2}') """.format(name,job,address) print(sql) # 执行sql cursor.execute(sql)def main(): faker = Faker("zh_CN") # 实例化faker args = args_list() # 获取用户输入的信息 #连接数据库 --> conn_db conn = conn_db(**dict(host=args.hostname,user=args.username,password=args.password,port=args.port)) # 创建游标 cursor = conn.cursor() # 创建表 --> create_table create_table(cursor) # 插入数据 insert_data(cursor,args.linenumber,faker)
if __name__ == '__main__': main()
#运行01[root@yunweisn testdemo]# python3 pymysql-test02.py --host ssh.51yunwei.top -utestuser -pTestUser@123 -n 5
insert into `test`.`test001`(name,job,address) values('陈秀兰','其他','河北省柳州市闵行东莞街H座 663475')
insert into `test`.`test001`(name,job,address) values('陈成','学徒工','海南省上海市华龙农路H座 901586')
insert into `test`.`test001`(name,job,address) values('金金凤','药品生产/质量管理','河南省莹县清浦崔街H座 341881')
insert into `test`.`test001`(name,job,address) values('刘斌','汽车修理工','台湾省桂花县涪城赵路e座 673426')
insert into `test`.`test001`(name,job,address) values('雷超','射频工程师','吉林省瑞市高港南宁路I座 464175')
# 登陆到数据库里看看MariaDB [test]> select * from test001;+----+-----------+---------------------------+----------------------------------------------+| id | name | job | address |+----+-----------+---------------------------+----------------------------------------------+| 1 | 陈秀兰 | 其他 | 河北省柳州市闵行东莞街H座 663475 || 2 | 陈成 | 学徒工 | 海南省上海市华龙农路H座 901586 || 3 | 金金凤 | 药品生产/质量管理 | 河南省莹县清浦崔街H座 341881 || 4 | 刘斌 | 汽车修理工 | 台湾省桂花县涪城赵路e座 673426 || 5 | 雷超 | 射频工程师 | 吉林省瑞市高港南宁路I座 464175 |+----+-----------+---------------------------+----------------------------------------------+5 rows in set (0.00 sec)0x06 加上多进程，小工具出炉！
#！/usr/bin/python3#conding=utf-8import argparseimport pymysqlfrom faker import Faker import sysimport threading
def args_list(): parse = argparse.ArgumentParser('connect to mysql tools') parse.add_argument("--host",dest='hostname',action='store',default='localhost',help='mysql server hostname. default localhost') parse.add_argument("-u","--user",dest='username',action='store',default='root',help='connect to mysql server user. default root') parse.add_argument("-p","--password",dest='password',action='store',help='connect to mysql server password.') parse.add_argument("-P","--port",dest='port',action='store',default=3306,type=int,help='mysql server port. default 3306') parse.add_argument("-n","--linenumber",dest='linenumber',default=500,type=int,help='insert data line , default 500') # 添加一个线程数量的选项 parse.add_argument("-t","--thread_size",dest='thread_size',default=5,type=int,help='thread number，default 5')
args = parse.parse_args() return args
def conn_db(**kwargs): try: conn = pymysql.connect(**kwargs) return conn except Exception as err: print("连接数据库错误,err=",err) sys.exit(1)
def create_table(cursor): sql = """ create table if not exists `test`.`test001` (id int(10) not null auto_increment, name varchar(255) not null, job varchar(255) not null, address varchar(255) not null, primary key (`id`)) character set = utf8 """ try: cursor.execute(sql) except Exception as err: print("创建表test001错误,err=",err) sys.exit(2)
def insert_data(conn_dict,linenumber,faker): # 为每个线程都创建一个conn，避免资源争用 conn = conn_db(**conn_dict) cursor = conn.cursor() for i in range(linenumber): name = faker.name() job = faker.job() address = faker.address() sql = """ insert into `test`.`test001`(name,job,address) values('{0}','{1}','{2}') """.format(name,job,address) print(sql) cursor.execute(sql) cursor.execute('commit') # 线程操作完成后提交并关闭链接 cursor.close() conn.close()def main(): faker = Faker("zh_CN") args = args_list() conn_dict = dict(host=args.hostname,user=args.username,password=args.password,port=args.port) conn = conn_db(**dict(host=args.hostname,user=args.username,password=args.password,port=args.port)) cursor = conn.cursor() create_table(cursor) # 创建表后关闭链接 cursor.close() conn.close()
threads = [] for i in range(args.thread_size): t = threading.Thread(target=insert_data,args=(conn_dict, args.linenumber,faker)) threads.append(t) t.start() for t in threads:# 进程阻塞，避免未执行完成主进程退出 t.join()
if __name__ == '__main__': main()
# 运行01[root@yunweisn testdemo]# python3 pymysql-test02.py --host localhost -uroot -proot -t 5 -n 500......# 查看结果MariaDB [test]> select count(*) from test001;+----------+| count(*) |+----------+| 2530 |+----------+1 row in set (0.00 sec)
[python运维] 使用python3制作一个mysql压测小工具！相关推荐
- 利用 C# 中的 FileSystemWatcher 制作一个文件夹监控小工具
利用 C# 中的 FileSystemWatcher 制作一个文件夹监控小工具 独立观察员 2020 年 12 月 26 日 前一段看到微信公众号 "码农读书" 上发了一篇文章&l ...
- Python：教你如何写一个测量网速的小工具
hello,大家好,我是wangzirui32,今天我们来学习如何写一个测量网速的小工具,开始学习吧! 1. pip安装 这个项目主要依赖speedtest_cli模块,pip安装: pip inst ...
- 制作一个每日一图小工具
如何制作一个发送每日一图的小工具? 目录 如何制作一个发送每日一图的小工具? 前言 第一步:获取图片 方法一:使用随机图片接口 方法二:从Pixiv网站获取好看的插图 第二步:存储图片 第三步:将图片 ...
- Python下借助百度翻译API制作一个翻译pdf的小工具-01
出于需要经常会读一些英语的pdf文档,奈何英语太差只得借助机翻.每次都需要把pdf里的文档复制粘贴到翻译软件里,接着在把结果复制到word文档里,之间还需要排版什么的.今天突然发现百度翻译有一个公开的 ...
- 使用python中的tkinter模块制作一个学习打卡小软件
学习任务打卡小软件---戈多Sensei 1.制定任务 2.完成任务 3.学习评估 最近为了给我的星星⭐写一个学习记录软件,学习了一下python中的tkinter模块,做了一个可视化打卡界面,此处为 ...
- python简单图画程序_制作一个简单的画图小程序界面
1 新建一个文本文档,然后点击重命名,建立一个python文件.后缀为.py.如图所示,系统会提示是否更改后缀,点击是.然后就创建了一个python文件. 2 打开刚才创建的文件,使用idle打开方式 ...
- python运维系统开发_Python系统运维开发实战
课程主题: Python 高级运维开发实战 课程讲师: Alex 老师, triaquae python 开源运维管理软件创始人,知名 IT 公司运维开发架构师 课程安排: 每周六一天全天(早 9:0 ...
- python运维脚本面试_运维开发工程师 面试题 shell编程
1. 32位随机密码生成 cat /proc/sys/kernel/random/uuid | tr -d '-' 2.查看当前系统每个ip的tcp连接数 -n 强制显示IP地址 -t 显示TCP连接 ...
- python程序员需要掌握哪些技术-python运维要掌握哪些内容
python运维需要会什么 随着移动互联网的普及,服务器运维所面临的挑战也随之越来越大.当规模增长到一定程度,手动管理方式已经无法应对,自动化运维成为解决问题的银弹. Python凭借其灵活性,在自动 ...
最新文章
- VirtualBox的四种网络连接方式
- CodeCombat编程游戏
- Flutter Beta 3 新特性概览
- python binascii array('c')_详解Python中的array数组模块相关使用
- PyQt4重写事件处理方法
- java 包错_以下关于Java包的描述中,错误的是()
- 性能测试：性能测试指标评估方法
- Springboot--Ehcache-Jpa （1）
- PHP 三种方式实现链式操作
- 使用windows内存-内存映射文件
- 用cmd命令下载百度云的资源
- Mac怎么看剩余空间，Mac怎么看硬盘空间
- 图片太大怎么压缩变小？图片如何压缩？
- 洛克菲勒写给儿子的38封信（上）、起点不决定终点、别让精神破产、我奋斗，我成功、我不依赖天赐的运气，但我靠策划运气平步青云、后退就是投降
- 如何为您的ADC选择最合适的基准电压源和放大器
- 2060显卡驱动最新版本_如何更新你的显卡驱动程序
- 计算机详细配置快捷键,Win7系统电脑快捷键设置大全
- 如何在小方框上打对号 小方框内打对勾 word 方框打对勾
- Android 12 行为变更，对应用产生的影响
- 振弦式传感器数据采集到水库大坝监测云平台进行监控和报警