(self, get)
| 3379 | |
| 3380 | # 获取数据库表 |
| 3381 | def GetDatabasesList(self, get): |
| 3382 | if not hasattr(get, "sid"): |
| 3383 | return public.returnMsg(False, "缺少参数！sid") |
| 3384 | if not str(get.sid).isdigit(): |
| 3385 | return public.returnMsg(False, "参数错误！sid") |
| 3386 | |
| 3387 | mysql_obj = public.get_mysql_obj_by_sid(get.sid) |
| 3388 | if not mysql_obj: return public.returnMsg(False, '数据库连接失败') |
| 3389 | data_list = mysql_obj.query("select schema_name from information_schema.schemata where schema_name not in ('sys', 'mysql', 'information_schema', 'performance_schema');") |
| 3390 | isError = self.IsSqlError(data_list) |
| 3391 | if isError != None: |
| 3392 | return isError |
| 3393 | if not isinstance(data_list, list): |
| 3394 | return public.returnMsg(False, "查询数据库失败！{}".format(data_list)) |
| 3395 | |
| 3396 | db_list = [{"name": "全局权限", "value": "*", "tb_list": []}] |
| 3397 | for data in data_list: |
| 3398 | db_name = data[0] |
| 3399 | info = { |
| 3400 | "name": db_name, |
| 3401 | "value": db_name, |
| 3402 | "tb_list": [{"name": "所有", "value": "*"}], |
| 3403 | } |
| 3404 | |
| 3405 | table_list = mysql_obj.query("show tables from `{db_name}`;".format(db_name=db_name)) |
| 3406 | if not isinstance(table_list, list): |
| 3407 | continue |
| 3408 | info["tb_list"].extend([{"name": data[0], "value": data[0]} for data in table_list]) |
| 3409 | |
| 3410 | db_list.append(info) |
| 3411 | |
| 3412 | return {"status": True, "msg": "ok", "data": db_list} |
| 3413 | |
| 3414 | # 获取所有用户 |
| 3415 | def GetMysqlUser(self, get): |
nothing calls this directly
no test coverage detected