提交代码

bin/DBConfig.py

@@ -0,0 +1,89 @@
+# -*- 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.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.29.9:3306/test2"
+    mysql_info_dict['db_user'] =  "root"
+    mysql_info_dict['db_password'] =  "Unary@2023"
+
+    #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` timestamp NULL DEFAULT NULL"
+    create_sql_string += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;"
+    mysql_info_dict['create_sql_string'] =  create_sql_string
+
+    # insert into sql
+    insert_list = []
+    data = faker_data.faker_data(lines=2)[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 2"
+    mysql_info_dict['del_sql'] =  del_sql
+
+
+    # uptata sql
+    updata_sql = "UPDATE "+ table_name + " t SET t.job='测试' WHERE t.name like '李%' AND  sex = 1 AND  id = 1 "
+    mysql_info_dict['updata_sql'] =  updata_sql
+
+    return(mysql_info_dict)
+
+
+def oracle_config():
+    mysql_info_dict={}
+    mysql_info_dict['driver_name'] =  "com.mysql.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.29.9:3306/test2"
+    mysql_info_dict['db_user'] =  "root"
+    mysql_info_dict['db_password'] =  "Unary@2023"
+    
+    return(mysql_info_dict)
+

bin/SQLDataCreating.py

@@ -1,24 +1,27 @@
 # -*- coding:utf-8 -*-
 import _load
 
-import faker_data
+import DBConfig
-import yaml
-import codecs
 
 import jaydebeapi
+
 from schedule import every, repeat, run_pending
-from pathlib import Path
-from datetime import datetime
 
 
-# 定义公共的部分
-path = Path(__file__)
-etc_dir = path.parent.parent / "etc"
-configfile = etc_dir / "test_config.yml"
-db_config = yaml.load(codecs.open(
-    configfile, 'r', encoding='utf-8'), Loader=yaml.FullLoader)
-driver_jar_path = etc_dir / "driver" / db_config["driver_jar"]
 
+#获取连接信息
+def get_db_info(DBType = "mysql"):
+    db_config = {}
+    if DBType == "mysql":
+        db_config = DBConfig.mysql_config()
+    elif DBType == "mysql":
+         db_config = DBConfig.oracle_config()
+    else:
+        assert False ,"DBType is not support."
+    return(db_config)
+
+#处理连接信息
+db_config = get_db_info()
 OAuth = [str(db_config["db_user"]), str(db_config["db_password"])]
 if not db_config["db_user"] and not db_config["db_password"]:
     OAuth = None
@@ -26,53 +29,52 @@ if not db_config["db_user"] and not db_config["db_password"]:
 conn = jaydebeapi.connect(db_config["driver_name"],
                           db_config["jdbc_url"],
                           OAuth,
-                          str(driver_jar_path))
+                          db_config["driver_jar_path"])
 # conn.close()
 
-
+#执行sql
 def execute_sql(sql):
     with conn.cursor() as curs:
         curs.execute(sql)
 
-# selecet_sql = "select * from t22 t  WHERE  name like '李%'"
-# execute_sql(selecet_sql)
-
 #@repeat(every(1).hours)
+#TODO(MH):按小时创建表，不太好控制，后面再去搞
 def create_table():
-    print("create_table ....")
+    print("start create table...")
-    sql_string = "CREATE TABLE `"
+    sql_string = db_config["create_sql_string"]
-    sql_string += "test"
-    sql_string += "` ("
-    sql_string += "`uuid` varchar(50) DEFAULT NULL,"
-    sql_string += "`id` bigint DEFAULT NULL,"
-    sql_string += "`name` varchar(50) DEFAULT NULL,"
-    sql_string += "`mobile` varchar(50) DEFAULT NULL,"
-    sql_string += "`ssn` varchar(50) DEFAULT NULL,"
-    sql_string += "`sex` int DEFAULT NULL,"
-    sql_string += "`email` varchar(50) DEFAULT NULL,"
-    sql_string += "`job` varchar(50) DEFAULT NULL,"
-    sql_string += "`address` varchar(50) DEFAULT NULL,"
-    sql_string += "`actime_time` timestamp NULL DEFAULT NULL"
-    sql_string += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;"
- 
     execute_sql(sql_string)
 
 
-@repeat(every(20).seconds)
+#每秒插入1条数据
+@repeat(every(2).seconds)
 def insert_data():
-    print("insert_data ....")
+    print("insert data ....")
+    #重新获取一下，确保每次的插入的数据都不一样
+    db_config2 = get_db_info()
+    insert_list = db_config2["insert_list"]
+    #执行插入
+    for sql_string in insert_list: 
+        execute_sql(sql_string)
 
 
-@repeat(every(5).seconds)
+#每5分钟删除100条数据
+@repeat(every(10).seconds)
 def dalete_data():
-    print("dalete_data....")
+    print("dalete data....")
+    sql_string = db_config["del_sql"]
+    execute_sql(sql_string)
 
 
-@repeat(every(6).seconds)
+#每1小时，修改一下数据
+@repeat(every(20).seconds)
 def updata_data():
-    print("updata_data ....")
+    print("updata data ....")
+    sql_string = db_config["updata_sql"]
+    execute_sql(sql_string)
+    
 
 
 if __name__ == '__main__':
+    create_table()
     while True:
         run_pending()

requirements.txt

@@ -1,5 +1,4 @@
 faker==19.12.0
 schedule==1.2.1
-PyYAML==6.0.1
 JayDeBeApi==1.2.3