修改数据库连接

conf/db.py

@@ -1,52 +0,0 @@
-# -*- coding: utf-8 -*-
-# @Time    : 2024/6/19
-# @Author  : 魏志亮
-
-mysql_config = \
-    {'plt_dev': {'database': 'energy',
-                 'host': '192.168.50.233',
-                 'password': 'admin123456',
-                 'port': 3306,
-                 'user': 'admin'},
-
-     'plt_prod': {'database': 'energy_prod',
-                  'host': '192.168.50.233',
-                  'password': 'admin123456',
-                  'port': 3306,
-                  'user': 'admin'},
-
-     'plt_connect_pool_config':
-         {'blocking': True,
-          'charset': 'utf8mb4',
-          'maxcached': 5,
-          'maxconnections': 20,
-          'maxshared': 0,
-          'mincached': 2,
-          'setsession': []},
-
-     'trans_dev': {'database': 'energy_data',
-                   'host': '192.168.50.235',
-                   'password': 'admin123456',
-                   'port': 30306,
-                   'user': 'root'},
-
-     'trans_prod': {'database': 'energy_data_prod',
-                    'host': '192.168.50.235',
-                    'password': 'admin123456',
-                    'port': 30306,
-                    'user': 'root'},
-
-     'trans_connect_pool_config':
-         {'blocking': True,
-          'charset': 'utf8',
-          'maxcached': 20,
-          'maxconnections': 20,
-          'maxshared': 0,
-          'mincached': 1,
-          'setsession': []}
-     }
-
-
-if __name__ == '__main__':
-    import yaml
-    print(yaml.dump(mysql_config, indent=4))

conf/etl_config_datang.yaml

@@ -0,0 +1,26 @@
+plt:
+  database: energy_prod
+  host: 172.16.37.22
+  password: admin123456
+  port: 3306
+  user: root
+trans:
+  database: energy_data_prod
+  host: 172.16.37.24
+  password: admin123456
+  port: 3306
+  user: root
+
+# 如果要放在原始路径,则配置这个 以下面的名称作为切割点,新建清理数据文件夹
+etl_origin_path_contain: 收资数据
+# 如果单独保存,配置这个路径
+save_path:
+
+log_path_dir: /data/collection_data/logs
+
+# 临时文件存放处,有些甲方公司隔得tmp太小,只好自己配置
+tmp_base_path: /data/collection_data/tmp
+
+run_batch_count: 1
+# 执行环境
+env: datang

conf/etl_config_dev.yaml

@@ -1,43 +1,16 @@
-plt_connect_pool_config:
-  blocking: true
-  charset: utf8mb4
-  maxcached: 5
-  maxconnections: 20
-  maxshared: 0
-  mincached: 2
-  setsession: [ ]
-plt_dev:
+plt:
   database: energy
   host: 192.168.50.233
   password: admin123456
   port: 3306
   user: admin
-plt_prod:
-  database: energy_prod
-  host: 192.168.50.233
-  password: admin123456
-  port: 3306
-  user: admin
-trans_connect_pool_config:
-  blocking: true
-  charset: utf8
-  maxcached: 20
-  maxconnections: 20
-  maxshared: 0
-  mincached: 1
-  setsession: [ ]
-trans_dev:
+
+trans:
   database: energy_data
   host: 192.168.50.235
   password: admin123456
   port: 30306
   user: root
-trans_prod:
-  database: energy_data_prod
-  host: 192.168.50.235
-  password: admin123456
-  port: 30306
-  user: root
 
 # 如果要放在原始路径,则配置这个 以下面的名称作为切割点,新建清理数据文件夹
 etl_origin_path_contain: 收资数据

conf/etl_config.yaml → conf/etl_config_prod.yaml

@@ -1,38 +1,11 @@
-plt_connect_pool_config:
-  blocking: true
-  charset: utf8mb4
-  maxcached: 5
-  maxconnections: 20
-  maxshared: 0
-  mincached: 2
-  setsession: [ ]
-plt_dev:
-  database: energy
-  host: 192.168.50.233
-  password: admin123456
-  port: 3306
-  user: admin
-plt_prod:
+plt:
   database: energy_prod
   host: 192.168.50.233
   password: admin123456
   port: 3306
   user: admin
-trans_connect_pool_config:
-  blocking: true
-  charset: utf8
-  maxcached: 20
-  maxconnections: 20
-  maxshared: 0
-  mincached: 1
-  setsession: [ ]
-trans_dev:
-  database: energy_data
-  host: 192.168.50.235
-  password: admin123456
-  port: 30306
-  user: root
-trans_prod:
+
+trans:
   database: energy_data_prod
   host: 192.168.50.235
   password: admin123456

utils/db/ConnectMysql.py

@@ -15,14 +15,11 @@ class ConnectMysql:
     def __init__(self, connet_name):
         self.yaml_data = yaml_conf(os.environ.get('ETL_CONF', "/data/config/etl_config.yaml"))
         self.connet_name = connet_name
-        if 'env' in os.environ:
-            self.env = os.environ['env']
-        else:
-            self.env = 'dev'
+        self.config = self.yaml_data[self.connet_name]
 
     # 从连接池中获取一个连接
     def get_conn(self):
-        return pymysql.connect(**self.yaml_data[self.connet_name + "_" + self.env])
+        return pymysql.connect(**self.config)
 
     # 使用连接执行sql
     def execute(self, sql, params=tuple()):
@@ -41,23 +38,18 @@ class ConnectMysql:
                     conn.rollback()
                     raise e
 
-    def execute_df_save(self, df, table_name):
-        config = self.yaml_data[self.connet_name + "_" + self.env]
+    def get_engine(self):
+        config = self.config
         username = config['user']
         password = config['password']
         host = config['host']
         port = config['port']
         dbname = config['database']
-        engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{dbname}')
-        df.to_sql(table_name, engine, index=False, if_exists='append')
+        return create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{dbname}')
+
+    def execute_df_save(self, df, table_name):
+        df.to_sql(table_name, self.get_engine(), index=False, if_exists='append')
 
     def read_sql_to_df(self, sql):
-        config = self.yaml_data[self.connet_name + "_" + self.env]
-        username = config['user']
-        password = config['password']
-        host = config['host']
-        port = config['port']
-        dbname = config['database']
-        engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{dbname}')
-        df = pd.read_sql_query(sql, engine)
+        df = pd.read_sql_query(sql, self.get_engine())
         return df