(self, get)
| 1245 | |
| 1246 | # 获取当前数据库信息 |
| 1247 | def GetInfo(self, get): |
| 1248 | db_name = get.db_name |
| 1249 | |
| 1250 | pgsql_obj = panelPgsql() |
| 1251 | db_find = public.M('databases').where("name=? AND LOWER(type)=LOWER('PgSql')", db_name).find() |
| 1252 | if not db_find: |
| 1253 | return public.returnMsg(False, '数据库不存在！') |
| 1254 | if db_find['sid']: |
| 1255 | db_find = public.M('database_servers').where("id=? AND LOWER(db_type)=LOWER('pgsql')", db_find['sid']).find() |
| 1256 | else: |
| 1257 | path = os.path.join(public.get_panel_path(), "data/postgresAS.json") |
| 1258 | if not os.path.isfile(path): |
| 1259 | return public.returnMsg(False, "请先设置管理员密码！") |
| 1260 | |
| 1261 | db_find = { |
| 1262 | "db_host": "127.0.0.1", |
| 1263 | "db_port": panelPgsql.get_config_options("port", int, 5432), |
| 1264 | "db_user": "postgres", |
| 1265 | "db_password": json.loads(public.readFile(path))['password'], |
| 1266 | } |
| 1267 | |
| 1268 | pgsql_obj.set_host(host=db_find['db_host'], port=db_find['db_port'], database=db_name, user=db_find['db_user'], password=db_find['db_password']) |
| 1269 | status, err_msg = pgsql_obj.connect() |
| 1270 | if status is False: |
| 1271 | return public.returnMsg(False, "连接数据库失败！") |
| 1272 | |
| 1273 | database_list = pgsql_obj.query(f"select datname from pg_database where datname='{db_name}';") |
| 1274 | database_list = [database[0] for database in database_list] |
| 1275 | if not database_list: |
| 1276 | return public.returnMsg(False, '数据库不存在！') |
| 1277 | |
| 1278 | collation = pgsql_obj.query(f"select pg_encoding_to_char(encoding) from pg_database where datname='{db_name}';")[0][0] |
| 1279 | result = { |
| 1280 | "database": db_name, |
| 1281 | "total_size": 0, |
| 1282 | "table_list": [], |
| 1283 | } |
| 1284 | |
| 1285 | temp_list = pgsql_obj.query(""" |
| 1286 | select schemaname, tablename |
| 1287 | from pg_tables |
| 1288 | where schemaname NOT IN ('pg_catalog', 'information_schema'); |
| 1289 | """) |
| 1290 | |
| 1291 | for table in temp_list: |
| 1292 | schema_name = table[0] |
| 1293 | table_name = table[1] |
| 1294 | full_table_name = f'"{schema_name}"."{table_name}"' # 注意加引号防止大小写或特殊字符问题 |
| 1295 | |
| 1296 | temp_data = pgsql_obj.query(f"select count(*) from {full_table_name};") |
| 1297 | if not isinstance(temp_data, list): continue |
| 1298 | if len(temp_data) == 0: continue |
| 1299 | rows_count = temp_data[0][0] |
| 1300 | |
| 1301 | data_size = pgsql_obj.query(f""" |
| 1302 | select pg_size_pretty(pg_table_size('{full_table_name}')) AS table_size, |
| 1303 | pg_size_pretty(pg_indexes_size('{full_table_name}')) AS indexes_size, |
| 1304 | pg_size_pretty(pg_total_relation_size('{full_table_name}')) AS total_size, |
no test coverage detected