(self, db_find: dict, args: dict)
| 1467 | |
| 1468 | # pgsql 备份数据库 |
| 1469 | def pgsql_backup_database(self, db_find: dict, args: dict) -> Tuple[bool, str]: |
| 1470 | from databaseModel.pgsqlModel import panelPgsql |
| 1471 | |
| 1472 | storage_type = args.get("storage_type", "db") # 备份的文件数量， 按照数据库 | 按照表 |
| 1473 | table_list = args.get("table_list", []) # 备份的集合 |
| 1474 | |
| 1475 | db_name = db_find["name"] |
| 1476 | isinstance |
| 1477 | db_user = "postgres" |
| 1478 | db_host = "127.0.0.1" |
| 1479 | if db_find["db_type"] == 0: |
| 1480 | db_port = panelPgsql.get_config_options("port", int, 5432) |
| 1481 | |
| 1482 | t_path = os.path.join(public.get_panel_path(), "data/postgresAS.json") |
| 1483 | if not os.path.isfile(t_path): |
| 1484 | error_msg = "管理员密码未设置！" |
| 1485 | self.echo_error(error_msg) |
| 1486 | return False, error_msg |
| 1487 | db_password = json.loads(public.readFile(t_path)).get("password", "") |
| 1488 | if not db_password: |
| 1489 | error_msg = "数据库密码为空！请先设置数据库密码！" |
| 1490 | self.echo_error(error_msg) |
| 1491 | return False, error_msg |
| 1492 | |
| 1493 | elif db_find["db_type"] == 1: |
| 1494 | # 远程数据库 |
| 1495 | conn_config = json.loads(db_find["conn_config"]) |
| 1496 | db_host = conn_config["db_host"] |
| 1497 | db_port = conn_config["db_port"] |
| 1498 | db_user = conn_config["db_user"] |
| 1499 | db_password = conn_config["db_password"] |
| 1500 | elif db_find["db_type"] == 2: |
| 1501 | conn_config = public.M("database_servers").where("id=? AND LOWER(db_type)=LOWER('pgsql')", db_find["sid"]).find() |
| 1502 | db_host = conn_config["db_host"] |
| 1503 | db_port = conn_config["db_port"] |
| 1504 | db_user = conn_config["db_user"] |
| 1505 | db_password = conn_config["db_password"] |
| 1506 | else: |
| 1507 | error_msg = "未知的数据库类型" |
| 1508 | self.echo_error(error_msg) |
| 1509 | return False, error_msg |
| 1510 | |
| 1511 | pgsql_obj = panelPgsql().set_host(host=db_host, port=db_port, database=db_name, user=db_user, password=db_password) |
| 1512 | status, err_msg = pgsql_obj.connect() |
| 1513 | if status is False: |
| 1514 | error_msg = "连接数据库[{}:{}]失败".format(db_host, int(db_port)) |
| 1515 | self.echo_error(error_msg) |
| 1516 | return False, error_msg |
| 1517 | |
| 1518 | db_size = 0 |
| 1519 | db_data = pgsql_obj.query("SELECT pg_database_size('{}') AS database_size;".format(db_name)) |
| 1520 | if isinstance(db_data, list) and len(db_data) != 0: |
| 1521 | db_size = db_data[0][0] |
| 1522 | |
| 1523 | if db_size == 0: |
| 1524 | error_msg = '指定数据库 `{}` 没有任何数据!'.format(db_name) |
| 1525 | self.echo_error(error_msg) |
| 1526 | return False, error_msg |
nothing calls this directly
no test coverage detected