database_connect_test.py 1.2 KB
# coding=utf-8
# author:        4N
# createtime:    2020/6/22
# email:         nheweijun@sina.com
from contextlib import closing

from sqlalchemy import create_engine

from app.models import DES,Database


from app.util.component.ApiTemplate import ApiTemplate
class Api(ApiTemplate):
    api_name = "测试数据库连接"
    def process(self):
        res = {}
        try:
            guid = self.para.get("guid")
            database = Database.query.filter_by(guid=guid)
            dbase:Database = database.one_or_none()
            engine = create_engine(DES.decode(dbase.sqlalchemy_uri), connect_args={'connect_timeout': 1})
            with closing(engine.connect()):
                pass
            res["result"]=True
            res["msg"] = "测试连接成功!"
        except:
            raise Exception("测试连接失败!")
        return res

    api_doc={
    "tags":["数据库接口"],
    "parameters":[
    
        {"name": "guid",
         "in": "formData",
         "type": "string", "required": "true"},
    
    
    ],
    "responses":{
        200:{
            "schema":{
                "properties":{
                }
            }
            }
        }
    }