database_edit.py 3.3 KB
# coding=utf-8
#author:        4N
#createtime:    2021/3/9
#email:         nheweijun@sina.com

from app.models import Database,db
from contextlib import closing

from sqlalchemy import create_engine

from app.models import DES
import datetime
from . import database_alias_check

from app.util.component.ApiTemplate import ApiTemplate

from app.util.component.PGUtil import PGUtil
class Api(ApiTemplate):
    api_name = "修改数据库"
    def process(self):
        
        res = {}
        res["result"] = False
        try:
            guid = self.para.get("guid")
            alias = self.para.get("alias")
    
            passwd = self.para.get("passwd")
            database = Database.query.filter_by(guid=guid)
            dbase:Database = database.one_or_none()

            update_dict={}
    
            if not dbase:
                res["msg"] = "数据库不存在!"
                return res
            if not alias and not passwd:
                res["msg"] = "请填写参数!"
                return res

    
            if alias:
                # 检测一波别名
                check_alias = database_alias_check.Api().result
                if not check_alias["result"]:
                    return check_alias
                update_dict["alias"]=alias
    
            if passwd:
                try:
                    user, password, host, port, database = PGUtil.get_info_from_sqlachemy_uri(DES.decode(dbase.sqlalchemy_uri))

                    sqlalchemy_uri = "postgresql://{}:{}@{}:{}/{}".format(user, passwd,host,
                                                                          port, database)

                    connectsrt = "hostaddr={} port={} dbname='{}' user='{}' password='{}'".format(host, port, database, user,
                                                                                                  passwd)
                    engine = create_engine(sqlalchemy_uri, connect_args={'connect_timeout': 2})
                    with closing(engine.connect()):
                        pass

                except :
                    res["msg"] = "密码错误!"
                    return res

                update_dict["sqlalchemy_uri"]= DES.encode(sqlalchemy_uri)
                update_dict["connectstr"] = DES.encode(connectsrt)
    
            if update_dict:
                update_dict["update_time"]=datetime.datetime.now()
                Database.query.filter_by(guid=guid).update(update_dict)
                db.session.commit()
            res["result"] = True
            res["msg"] = "数据修改成功!"
        except Exception as e:
            db.session.rollback()
            raise e
        return res
    
    
    
    api_doc={
    "tags":["数据库接口"],
    "parameters":[
        {"name": "guid",
         "in": "formData",
         "type": "string"},
        {"name": "alias",
         "in": "formData",
         "type": "string","description":"数据库别名"},
        {"name": "passwd",
         "in": "formData",
         "type": "string", "description": "数据库密码"}
    ],
    "responses":{
        200:{
            "schema":{
                "properties":{
                }
            }
            }
        }
    }