database_list.py 2.9 KB
# coding=utf-8
#author:        4N
#createtime:    2021/3/9
#email:         nheweijun@sina.com
from app.models import Database,DES
from contextlib import closing
from sqlalchemy import create_engine

from app.models import Database


from app.util.component.ApiTemplate import ApiTemplate
from app.util.component.ModelVisitor import ModelVisitor
class Api(ApiTemplate):
    api_name = "获取数据库列表"
    def process(self):
        res = {}
        res["data"]={}
        try:
            page_index = int(self.para.get("pageIndex", "0"))
            page_size = int(self.para.get("pageSize", "10"))
            alias = self.para.get("alias")
            database_guid = self.para.get("guid")
            database = Database.query.order_by(Database.create_time.desc())
    
            if alias:
                database = database.filter(Database.alias.like("%" + alias + "%"))
            if database_guid:
                database = database.filter_by(guid = database_guid)

            res["data"]["count"] = database.count()
    
            database = database.limit(page_size).offset(page_index * page_size).all()
            res["data"]["list"] =[]
            for datab in database:
                # 测试连接
                try:
                    engine = create_engine(DES.decode(datab.sqlalchemy_uri), connect_args={'connect_timeout': 1})
                    with closing(engine.connect()):
                        pass
                    available=1
                except:
                    available=-1
                table_count = datab.relate_tables.count()
                datab_json = ModelVisitor.database_to_json(datab)
                datab_json["table_count"] = table_count
                datab_json["available"] = available
                res["data"]["list"].append(datab_json)

            # 可用非可用排序
            sorted_list = [dat for dat in res["data"]["list"] if dat["available"]==1]
            sorted_list.extend([dat for dat in res["data"]["list"] if dat["available"] == -1])
            res["data"]["list"] = sorted_list
            res["result"]=True
        except Exception as e:
            raise e
        return res
    
    
    
    api_doc={
    "tags":["数据库接口"],
    "parameters":[
        {"name": "page_index",
         "in": "formData",
         "type": "int",
         "default": "0"},
        {"name": "page_size",
         "in": "formData",
         "type": "int",
         "default": "10"},
        {"name": "alias",
         "in": "formData",
         "type": "string","description":"数据库别名"},
        {"name": "guid",
         "in": "formData",
         "type": "string", "description": "数据库guid"}
    ],
    "responses":{
        200:{
            "schema":{
                "properties":{
                }
            }
            }
        }
    }