dataCreating/bin/DBConfig.py

# -*- 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 = path.parent.parent / "test_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['DATABASES']
    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 " + 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)