(self)
| 67 | |
| 68 | # 连接PGSQL数据库 |
| 69 | def connect(self) -> Tuple[bool, str]: |
| 70 | if self.__CONN_KWARGS.get("password") is None: |
| 71 | if self.__CONN_KWARGS["host"] in ["localhost", "127.0.0.1"] and os.path.exists("/www/server/pgsql/data"): |
| 72 | self.__CONN_KWARGS["port"] = self.get_config_options("port", int, 5432) |
| 73 | tmp_args = public.dict_obj() |
| 74 | tmp_args.is_True = True |
| 75 | self.__CONN_KWARGS["password"] = main().get_root_pwd(tmp_args) |
| 76 | |
| 77 | try: |
| 78 | self.__DB_CONN = psycopg2.connect(**self.__CONN_KWARGS) |
| 79 | self.__DB_CONN.autocommit = True |
| 80 | self.__DB_CONN.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) # <-- ADD THIS LINE |
| 81 | |
| 82 | self.__DB_CUR = self.__DB_CONN.cursor() |
| 83 | return True, "正常" |
| 84 | except: |
| 85 | err_msg = public.get_error_info() |
| 86 | return False, err_msg |
| 87 | |
| 88 | # 设置连接参数 |
| 89 | def set_host(self, *args, **kwargs): |
no test coverage detected