132 lines
4.7 KiB
Python
132 lines
4.7 KiB
Python
# -*- coding:utf-8 -*-
|
|
import faker_data
|
|
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
|
|
# 定义公共的部分
|
|
path = Path(__file__)
|
|
etc_dir = path.parent.parent / "etc"
|
|
|
|
|
|
# def get_now():
|
|
# now = datetime.now()
|
|
# #now_day = now.strftime("%Y-%m-%d")
|
|
# now_hours = now.strftime("%Y%m%d%H")
|
|
# return(now_hours)
|
|
|
|
|
|
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://10.10.14.69:3306/sys"
|
|
mysql_info_dict['db_user'] = "root"
|
|
mysql_info_dict['db_password'] = "111111"
|
|
|
|
#create table sql
|
|
# now_hours = get_now()
|
|
#TODO(MH):根据时间生成不同的表
|
|
table_name = "mysql_test_table"
|
|
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=200)[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:@10.10.29.10:1521/unary"
|
|
oracle_info_dict['db_user'] = "Test01"
|
|
oracle_info_dict['db_password'] = "test123"
|
|
|
|
#create table sql
|
|
# now_hours = get_now()
|
|
#TODO(MH):根据时间生成不同的表
|
|
table_name = "ORACLE_TEST"
|
|
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=200)[1:]
|
|
for a in data:
|
|
#拼接sql
|
|
insert_sql_string = "INSERT INTO " + 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 "+ table_name + " WHERE rownum<=100"
|
|
oracle_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 "
|
|
oracle_info_dict['updata_sql'] = updata_sql
|
|
|
|
return(oracle_info_dict)
|
|
|