@name 检查远程数据库信息是否正确 @author hwliang<2022-01-06> @param conn_config 连接信息 db_host 服务器地址 db_port 数据库端口 db_user 用户名 db_name 数据库名称 db_password 数
(self, conn_config)
| 575 | return public.returnMsg(False, '添加失败： {}'.format(result)) |
| 576 | |
| 577 | def CheckCloudDatabase(self, conn_config): |
| 578 | ''' |
| 579 | @name 检查远程数据库信息是否正确 |
| 580 | @author hwliang<2022-01-06> |
| 581 | @param conn_config<dict> 连接信息 |
| 582 | db_host<string> 服务器地址 |
| 583 | db_port<port> 数据库端口 |
| 584 | db_user<string> 用户名 |
| 585 | db_name<string> 数据库名称 |
| 586 | db_password<string> 数据库密码 |
| 587 | @return True / dict |
| 588 | ''' |
| 589 | try: |
| 590 | if not 'db_name' in conn_config: conn_config['db_name'] = None |
| 591 | mysql_obj = db_mysql.panelMysql() |
| 592 | flag = mysql_obj.set_host(conn_config['db_host'], conn_config['db_port'], conn_config['db_name'], conn_config['db_user'], conn_config['db_password']) |
| 593 | if flag is False: |
| 594 | if mysql_obj._ex: |
| 595 | return public.returnMsg(False, self.GetMySQLError(mysql_obj._ex)) |
| 596 | return public.returnMsg(False, "连接数据库[{}:{}]失败".format(conn_config['db_host'], conn_config['db_port'])) |
| 597 | |
| 598 | result = mysql_obj.query("show databases") |
| 599 | if isinstance(result, str) or self.IsSqlError(result): |
| 600 | if mysql_obj._ex: |
| 601 | return public.returnMsg(False, self.GetMySQLError(mysql_obj._ex)) |
| 602 | else: |
| 603 | return public.returnMsg(False, self.GetMySQLError(result)) |
| 604 | if not conn_config['db_name']: return True |
| 605 | for i in result: |
| 606 | if i[0] == conn_config['db_name']: |
| 607 | return True |
| 608 | return public.returnMsg(False, '指定数据库不存在!') |
| 609 | except Exception as ex: |
| 610 | res = self.GetMySQLError(ex) |
| 611 | if not res: res = str(ex) |
| 612 | return public.returnMsg(False, res) |
| 613 | |
| 614 | def GetMySQLError(self, e): |
| 615 | if isinstance(e, str): |
no test coverage detected