Python 自动生成数据插入到表,只需要提供连接参数、库名、表名、与想要插入数据的数量
目录
1、前言
2、思路
2、代码
1、前言
作为一个测试人员，经常会有需要做测试数据的时候，所以一直想写这么一个demo分享出来，
2、思路
平常我们插入数据，需要慢吞吞的写一条insert语句，久而久之，因为插数据之事浪费了大量的时间，有时插入的数据基本相同，不能达到预期的测试效果。
之前也写过一篇文章，自动插入数据的，还是需要自己先知道表里所有的字段属性，只算是半自动化。
相对那篇来说，本篇算是全自动化了。
1、自动获取表字段属性，
2、通过属性判断字段属性自动生成数据集，
3、最后拼装成sql插入语句自动插入数据。
2、代码
使用到的库很少，全部都是Python的基础知识
import pymysql #连接数据库用import random #自动生成数据用import copy #深度copy用import datetime #生成时间数据用class AutoFactory(object): def __init__(self): pass def connect_db(self, host, port, user, pwd, db): """ 连接数据库 :param host: :param port: :param user: :param pwd: :param db: :return: """ try: self.db = pymysql.connect(host=host, user=user, password=pwd, port=port, database=db, use_unicode=True, charset="utf8") return True except Exception: return False def decide_dbtabcolum(self, db, table): """ 取得表字段名与类型长度等参数，格式化成一个JSON字典 :param db: :param table: :return: """ cursor = self.db.cursor() sql = 'SELECT TABLE_NAME 表名,COLUMN_NAME 列名,COLUMN_TYPE 数据类型,DATA_TYPE 字段类型,CHARACTER_MAXIMUM_LENGTH 长度,IS_NULLABLE ' \'是否为空,COLUMN_DEFAULT 默认值,COLUMN_COMMENT 备注, EXTRA 是否自动增长, COLUMN_KEY 键类型 FROM INFORMATION_SCHEMA.COLUMNS where table_schema = {}' \' AND table_name = {}'.format(self.madevachar(db), self.madevachar(table)) cursor.execute(sql) values = cursor.fetchall() des = cursor.description list_des = [] for d in des: list_des.append(d[0]) list_re = [] for val in values: list_re.append(dict(zip(list_des, list(val)))) cursor.close() return list_re def madevachar(self, par): """ 格式化sql值 :param par: :return: """ return "'" + par + "'" def made_data(self, list_res, count=10): """ 自动生成数据,暂时实现了一部分数据类型， :param list_res: :param count: :return: list_all = [[],[],[]] """ list_all = [] for res in list_res: if res['字段类型'] == 'int': if ('id' in str(res['列名']).lower() or 'no' in str(res['列名']).lower()) and res['是否自动增长'] == 'auto_increment': print('id是自动生成的不需要数据') pass elif ('id' in str(res['列名']).lower() or 'no' in str(res['列名']).lower()) and res['键类型'] == 'PRI': sql = 'select {} from {} order by {} desc limit 1'.format(res['列名'], res['表名'], res['列名']) cursor = self.db.cursor() cursor.execute(sql) values = cursor.fetchall() max_id = values[0][0] list_id = [] for i in range(count): max_id += 1 list_id.append(max_id) list_all.append(copy.deepcopy(list_id)) list_id.clear() elif 'status' in res['列名']: a = res['长度'] > 1 and 19, 9 list_staus =[] list_s = [] if res['备注'] is not None: for i in range(a):if str(i) in res['备注']: list_staus.append(i) else: for i in range(a):list_staus.append(i) for j in range(count): list_s.append(random.choice(list_staus)) list_all.append(copy.deepcopy(list_s)) list_s.clear() elif 'date' in res['备注'] or 'time' in res['备注']: pass elif res['字段类型'] == 'varchar': if 'mobile' in str(res['列名']).lower() or 'phone' in str(res['列名']).lower(): list_phone = [] list_head = ['135', '132', '147', '153', '136', '138'] for i in range(count): head = random.choice(list_head) phone = head + ''.join(random.sample(['1','2','3','4','5','6','7','8','9','0'], 8)) list_phone.append(self.madevachar(phone)) list_all.append(copy.deepcopy(list_phone)) list_phone.clear() elif 'date' in str(res['列名']).lower() or 'time' in str(res['列名']): pass elif 'email' in str(res['列名']).lower(): list_email = [] list_end = ['@163.com', '@qq.com', '@126.com', '@gogle.com', '@aliyun.com'] for i in range(count): end = random.choice(list_end) email_ddr = ''.join(random.sample(['a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w', 'x','y','z','1','2','4','5','6','7','8','9','0','A','B','C','D','E','F','G','H','I','J','K', 'L','M','N','O','P','Q','R','S','T','U','V','W','X','Y','Z'], 10)) + end list_email.append(self.madevachar(email_ddr)) list_all.append(copy.deepcopy(list_email)) list_email.clear() else: list_par = [] for j in range(count): if res['长度'] > 61:i = random.randint(1, 61) else:i = random.randint(1, res['长度']) par = ''.join(random.sample(['a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w', 'x','y','z','1','2','4','5','6','7','8','9','0','A','B','C','D','E','F','G','H','I','J','K', 'L','M','N','O','P','Q','R','S','T','U','V','W','X','Y','Z'], i)) list_par.append(self.madevachar(par)) list_all.append(copy.deepcopy(list_par)) list_par.clear() elif res['字段类型'] == 'tinyint': print('tinyint暂时没写') elif res['字段类型'] == 'smallint': print('smallint暂时没写') elif res['字段类型'] == 'mediumint': print('mediumint暂时没写') elif res['字段类型'] == 'bigint': print('bigint暂时没写') elif res['字段类型'] == 'float': print('float暂时没写') elif res['字段类型'] == 'double': print('double暂时没写') elif res['字段类型'] == 'char': print('char暂时没写') elif res['字段类型'] == 'tinytext': print('tinytext暂时没写') elif res['字段类型'] == 'mediumtext': print('mediumtext暂时没写') elif res['字段类型'] == 'longtext': print('longtext暂时没写') elif res['字段类型'] == 'date': list_date = [] for i in range(count): now = datetime.date.today() dy = random.randint(1000, 5000) delta = datetime.timedelta(days=dy) day = now - delta list_date.append(self.madevachar(day.isoformat())) list_all.append(copy.deepcopy(list_date)) list_date.clear() elif res['字段类型'] == 'time': print('time暂时没写') elif res['字段类型'] == 'datetime': print('datetime暂时没写') elif res['字段类型'] == 'timestamp': print('timestamp暂时没写') return list_all def auto_isert(self, db, table, count=10): """ 取得参数与插入数据，拼装sql语句，插入数据 :param db: :param table: :param count: :return: """ list_res = self.decide_dbtabcolum(db=db, table=table) list_all = self.made_data(list_res=list_res, count=count) cursor = self.db.cursor() sql = 'insert into {} ('.format(table) for res in list_res: if res['是否自动增长'] != 'auto_increment': sql = sql + res['列名'] + ',' sql = sql.rstrip(sql[-1:]) + ') values(' try: for i in range(count): enpsql = sql for all in list_all: enpsql += str(all[i]) + ',' enpsql = enpsql.rstrip(enpsql[-1:]) + ');' cursor.execute(enpsql) cursor.close() self.db.commit() print('插入数据成功，哦耶！') except Exception as e: print('插入数据发生异常:[{}]'.format(str(e))) self.db.rollback() def insert_db(self, host, user, pwd, port, db, table, count): """ 入口 :param host: db服务器 :param user: db用户名 :param pwd: 密码 :param port: 端口 :param db: 数据库名 :param table: 表名 :param count: 插入数据的数量 :return: """ if self.connect_db(host=host, user=user, pwd=pwd, port=port, db=db): self.auto_isert(db=db, table=table, count=count) self.db.close() else: print('connect DB Filed, 请检查传入的数据库参数')"调用"auto = AutoFactory()auto.insert_db(host="192.168.10.37", user="root", pwd="qwer1234", port=3306, db="StudentV4DB", table="Student", count=10)