数据库连接测试超时问题修改

app/utils/common_tools.py

@@ -7,6 +7,49 @@ from huansi_utils.app.apploader import logger
 from huansi_utils.exception.exception import HSException
 
 
+# 超过四秒，基本上没救了，告辞~~~~
+@timeout_decorator.timeout(2)
+def test_sql_connection(ip, port, user_name, password, db_name):
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    con_str = f'mssql+pymssql://{user_name}:{password}@{ip}:{port}/{db_name}'
+    engine = create_engine(con_str)
+    DBSession = sessionmaker(bind=engine)
+    try:
+        session = DBSession()
+
+        result = session.execute("select test='huansi'")
+
+        data = result.fetchone()
+
+        if data.test == 'huansi':
+            return True
+        else:
+            return False
+
+    except Exception as e:
+        if isinstance(e, timeout_decorator.TimeoutError):
+            logger.error("数据库连接失败，失败原因：{}".format(e))
+            return False
+        orig = getattr(e, 'orig')
+        args = getattr(orig, 'args', ((),))[0]
+        error_message = ''
+        for e_item in args:
+            if isinstance(e_item, bytes):
+                try:
+                    e_item = e_item.decode()
+                except:
+                    e_item = str(e_item)
+            error_message += str(e_item)
+
+        if not error_message:
+            error_message = str(e)
+
+        logger.error("数据库连接失败，失败原因：{}".format(error_message))
+
+        return False
+
+
 class CommonTools:
     @classmethod
     def validate_dict_value_all_not_none(cls, dict_data: dict, mapping=None):
@@ -21,9 +64,7 @@ class CommonTools:
             if value is None:
                 raise HSException(f"{key}不能为空")
 
-    # 超过四秒，基本上没救了，告辞~~~~
-    @timeout_decorator.timeout(2)
-    def test_db_connection(cls, ip: int, port: int, user_name: str, password: str, db_name: str):
+    def test_db_connection(self, ip: int, port: int, user_name: str, password: str, db_name: str):
         '''
         测试数据库连接
         :param ip:
@@ -34,48 +75,7 @@ class CommonTools:
         :param sql:
         :return:
         '''
-        from sqlalchemy import create_engine
-        from sqlalchemy.orm import sessionmaker
-
-        con_str = f'mssql+pymssql://{user_name}:{password}@{ip}:{port}/{db_name}'
-
-        engine = create_engine(con_str)
-
-        DBSession = sessionmaker(bind=engine)
-
-        try:
-            session = DBSession()
-
-            result = session.execute("select test='huansi'")
-
-            data = result.fetchone()
-
-            if data.test == 'huansi':
-                return True
-            else:
-                return False
-
-        except Exception as e:
-            if isinstance(e, TimeoutError):
-                logger.error("数据库连接失败，失败原因：{}".format(e))
-                return False
-            orig = getattr(e, 'orig')
-            args = getattr(orig, 'args', ((),))[0]
-            error_message = ''
-            for e_item in args:
-                if isinstance(e_item, bytes):
-                    try:
-                        e_item = e_item.decode()
-                    except:
-                        e_item = str(e_item)
-                error_message += str(e_item)
-
-            if not error_message:
-                error_message = str(e)
-
-            logger.error("数据库连接失败，失败原因：{}".format(error_message))
-
-            return False
+        return test_sql_connection(ip, port, user_name, password, db_name)
 
     @classmethod
     def validate_runner_is_install(self):