因需要生成数据,验证数据库相关问题,且此需求可能以后也会经常遇到,因此分享出来~
在基础的云ECS,通过python3执行即可
- # encoding=utf-8
- import pymysql
- import random
- import time
-
-
- # create table student_source(id int NOT NULL PRIMARY KEY AUTO_INCREMENT,name varchar(20),subject varchar(20),score int(5))
-
- class RandomDataInsert:
- # 初始化类
- def __init__(self, ty):
- self.type = ty
- self.coon = None
- self.sqlList = []
- pass
-
- # 连接数据库
- def connectDB(self, ip, port, dbname, user=None, pwd=None, ):
- if self.type == 'mysql':
- self.coon = pymysql.connect(
- host=ip,
- user=user,
- passwd=pwd,
- port=port,
- db=dbname,
- charset='utf8'
- )
- return True
-
- # 关闭连接
- def disconnectDB(self):
- self.coon.close()
-
- # 添加sql语句
- def appendSql(self, sql):
- if sql is not None:
- self.sqlList.append(sql)
-
- # 执行SQL
- def executeInsertAll(self):
- try:
- cur = self.coon.cursor()
- for sql in self.sqlList:
- print("Going to execute: ", sql)
- cur.execute(sql)
- self.coon.commit()
- cur.close()
- except Exception as e:
- print('发生异常', e)
- self.coon.rollback()
-
-
- # 生成数据SQL小工厂
- def defineSql(ope, count):
- for i in range(count):
- subject = random.choice(['Chinese', 'Math', 'English'])
- score = random.randint(0, 100)
- name = ''.join(random.sample(
- ['z', 'y', 'x', 'w', 'v', 'u', 't', 's', 'r', 'q', 'p', 'o', 'n', 'm', 'l', 'k', 'j', 'i', 'h', 'g', 'f',
- 'e', 'd', 'c', 'b', 'a'], 2))
- sql = "insert into student values(null,'" + name + "','" + subject + "'," + str(score) + ");"
- ope.appendSql(sql)
-
-
- # 主函数
- if __name__ == '__main__':
- print('Random DB Insert begin!!')
- sqlList = RandomDataInsert('mysql')
- sqlList.connectDB('实例名称', 3306, '数据库', '账户', '密码')
- print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
- defineSql(sqlList, 20)
- print("defineSql,sqlList:{}", len(sqlList.sqlList))
- sqlList.executeInsertAll()
- sqlList.disconnectDB()