137 lines
5.0 KiB
Python
137 lines
5.0 KiB
Python
# -*- coding:utf-8 -*-
|
|
import faker_data
|
|
import yaml
|
|
|
|
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
|
|
# 定义公共的部分
|
|
path = Path(__file__)
|
|
etc_dir = path.parent.parent / "etc"
|
|
config_file_patch = etc_dir / "db_config.yml"
|
|
|
|
|
|
# 获取配置文件
|
|
with open(config_file_patch, "r", encoding='utf-8') as fy:
|
|
config = yaml.safe_load(fy)
|
|
# print(config)
|
|
|
|
|
|
def get_now():
|
|
week = str(datetime.now().isocalendar()[1])
|
|
return(week)
|
|
|
|
|
|
def mysql_config():
|
|
mysql_info_dict={}
|
|
mysql_info_dict['driver_name'] = "com.mysql.cj.jdbc.Driver"
|
|
mysql_info_dict['driver_jar'] = "mysql-connector-java-8.0.29.jar"
|
|
driver_jar_path = etc_dir / "driver" / mysql_info_dict['driver_jar']
|
|
mysql_info_dict['driver_jar_path'] = str(driver_jar_path)
|
|
#这里要连接到具体的库
|
|
mysql_info_dict['jdbc_url'] = "jdbc:mysql://" + config['IP'] + ":" + config['PROT'] + "/" + config['DATABASES']
|
|
mysql_info_dict['db_user'] = config['DBUSER']
|
|
mysql_info_dict['db_password'] = config['DBPASSWD']
|
|
|
|
#create table sql
|
|
week = get_now()
|
|
table_name = "mysql_test_table_w" + week
|
|
create_sql_string = "CREATE TABLE IF NOT EXISTS `"
|
|
create_sql_string += table_name
|
|
create_sql_string += "` ("
|
|
create_sql_string += "`uuid` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`id` bigint DEFAULT NULL,"
|
|
create_sql_string += "`name` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`mobile` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`ssn` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`sex` int DEFAULT NULL,"
|
|
create_sql_string += "`email` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`job` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`address` varchar(50) DEFAULT NULL,"
|
|
create_sql_string += "`actime_time` varchar(50) NULL DEFAULT NULL"
|
|
create_sql_string += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"
|
|
mysql_info_dict['create_sql_string'] = create_sql_string
|
|
|
|
# insert into sql
|
|
insert_list = []
|
|
data = faker_data.faker_data(lines=config['InsertRows'])[1:]
|
|
for a in data:
|
|
#拼接sql
|
|
insert_sql_string = "INSERT INTO " + table_name
|
|
insert_sql_string += "(uuid, id, name, mobile, ssn, sex, email, job, address, actime_time) VALUES("
|
|
str_data = ','.join("'{0}'".format(x) for x in a)
|
|
insert_sql_string += str_data + ");"
|
|
|
|
insert_list.append(insert_sql_string)
|
|
# 返回所有的插入语句
|
|
mysql_info_dict['insert_list'] = insert_list
|
|
|
|
# delete sql
|
|
del_sql = "delete FROM "+ table_name + " where 1=1 limit 100"
|
|
mysql_info_dict['del_sql'] = del_sql
|
|
|
|
|
|
# uptata sql
|
|
updata_sql = "UPDATE "+ table_name + " t SET t.job='unary_测试' WHERE t.name like '李%' AND sex = 1 AND id = 1 "
|
|
mysql_info_dict['updata_sql'] = updata_sql
|
|
|
|
return(mysql_info_dict)
|
|
|
|
|
|
def oracle_config():
|
|
oracle_info_dict={}
|
|
oracle_info_dict['driver_name'] = "oracle.jdbc.OracleDriver"
|
|
oracle_info_dict['driver_jar'] = "ojdbc8-12.2.0.1.jar"
|
|
driver_jar_path = etc_dir / "driver" / oracle_info_dict['driver_jar']
|
|
oracle_info_dict['driver_jar_path'] = str(driver_jar_path)
|
|
#这里要连接到具体的实例
|
|
oracle_info_dict['jdbc_url'] = "jdbc:oracle:thin:@" + config['IP'] + ":" + config['PROT'] + "/" + config['SERVER']
|
|
oracle_info_dict['db_user'] = config['DBUSER']
|
|
oracle_info_dict['db_password'] = config['DBPASSWD']
|
|
|
|
#create table sql
|
|
week = get_now()
|
|
table_name = "ORACLE_TEST_w" + week
|
|
create_sql_string = 'CREATE TABLE "'
|
|
create_sql_string += table_name + '" '
|
|
create_sql_string += '("UUID" VARCHAR2(50),'
|
|
create_sql_string += '"ID" NUMBER(38,0), '
|
|
create_sql_string += '"NAME" VARCHAR2(50),'
|
|
create_sql_string += '"MOBILE" VARCHAR2(50),'
|
|
create_sql_string += '"SSN" VARCHAR2(50),'
|
|
create_sql_string += '"SEX" NUMBER(*,0), '
|
|
create_sql_string += '"EMAIL" VARCHAR2(50), '
|
|
create_sql_string += '"JOB" VARCHAR2(50),'
|
|
create_sql_string += '"ADDRESS" VARCHAR2(100),'
|
|
create_sql_string += '"ACTIME_TIME" VARCHAR2(50))'
|
|
oracle_info_dict['create_sql_string'] = create_sql_string
|
|
|
|
|
|
# insert into sql
|
|
insert_list = []
|
|
data = faker_data.faker_data(lines=config['InsertRows'])[1:]
|
|
for a in data:
|
|
#拼接sql
|
|
insert_sql_string = 'INSERT INTO ' + config['DATABASES'] + '."' + table_name + '" '
|
|
insert_sql_string += " t (uuid, id, name, mobile, ssn, sex, email, job, address, actime_time) VALUES("
|
|
str_data = ','.join("'{0}'".format(x) for x in a)
|
|
insert_sql_string += str_data + ")"
|
|
|
|
insert_list.append(insert_sql_string)
|
|
# 返回所有的插入语句
|
|
oracle_info_dict['insert_list'] = insert_list
|
|
|
|
# delete sql
|
|
del_sql = 'delete FROM '+ config['DATABASES'] + '."' + table_name + '" WHERE rownum<=100'
|
|
oracle_info_dict['del_sql'] = del_sql
|
|
|
|
|
|
# uptata sql
|
|
updata_sql = "UPDATE "+ config['DATABASES'] + '."' + table_name + '" '+ " t SET t.job='unary_测试' WHERE t.name like '李%' AND sex = 1 AND id = 1 "
|
|
oracle_info_dict['updata_sql'] = updata_sql
|
|
|
|
return(oracle_info_dict)
|
|
|