Commit 7e9602bf authored by 金凯强's avatar 金凯强 🎨

还原上一次修改

parent 2219f038
...@@ -7,14 +7,42 @@ from huansi_utils.app.apploader import logger ...@@ -7,14 +7,42 @@ from huansi_utils.app.apploader import logger
from huansi_utils.exception.exception import HSException from huansi_utils.exception.exception import HSException
# 超过四秒,基本上没救了,告辞~~~~ class CommonTools:
@timeout_decorator.timeout(2) @classmethod
def test_sql_connection(ip, port, user_name, password, db_name): def validate_dict_value_all_not_none(cls, dict_data: dict, mapping=None):
'''
验证字典的值全不为空
:return:
'''
if not isinstance(dict_data, dict):
raise HSException('dict_data必须为字典类型')
for key, value in dict_data.items():
if value is None:
raise HSException(f"{key}不能为空")
# 超过四秒,基本上没救了,告辞~~~~
@timeout_decorator.timeout(2)
def test_db_connection(self, ip: int, port: int, user_name: str, password: str, db_name: str):
'''
测试数据库连接
:param ip:
:param port:
:param user_name:
:param password:
:param db_name:
:param sql:
:return:
'''
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
con_str = f'mssql+pymssql://{user_name}:{password}@{ip}:{port}/{db_name}' con_str = f'mssql+pymssql://{user_name}:{password}@{ip}:{port}/{db_name}'
engine = create_engine(con_str) engine = create_engine(con_str)
DBSession = sessionmaker(bind=engine) DBSession = sessionmaker(bind=engine)
try: try:
session = DBSession() session = DBSession()
...@@ -49,34 +77,6 @@ def test_sql_connection(ip, port, user_name, password, db_name): ...@@ -49,34 +77,6 @@ def test_sql_connection(ip, port, user_name, password, db_name):
return False return False
class CommonTools:
@classmethod
def validate_dict_value_all_not_none(cls, dict_data: dict, mapping=None):
'''
验证字典的值全不为空
:return:
'''
if not isinstance(dict_data, dict):
raise HSException('dict_data必须为字典类型')
for key, value in dict_data.items():
if value is None:
raise HSException(f"{key}不能为空")
def test_db_connection(self, ip: int, port: int, user_name: str, password: str, db_name: str):
'''
测试数据库连接
:param ip:
:param port:
:param user_name:
:param password:
:param db_name:
:param sql:
:return:
'''
return test_sql_connection(ip, port, user_name, password, db_name)
@classmethod @classmethod
def validate_runner_is_install(self): def validate_runner_is_install(self):
''' '''
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment