| 31 | |
| 32 | |
| 33 | class panelPgsql: |
| 34 | _CONFIG_PATH = os.path.join(public.get_setup_path(), "pgsql/data/postgresql.conf") |
| 35 | |
| 36 | def __init__(self): |
| 37 | self.check_package() |
| 38 | |
| 39 | self.__CONN_KWARGS = { |
| 40 | "host": "localhost", |
| 41 | "port": 5432, |
| 42 | "user": "postgres", |
| 43 | "password": None, |
| 44 | "database": None, |
| 45 | "connect_timeout": 3, |
| 46 | } |
| 47 | self.__DB_CONN = None |
| 48 | self.__DB_CUR = None |
| 49 | |
| 50 | # 检查python包是否存在 |
| 51 | @classmethod |
| 52 | def check_package(cls): |
| 53 | """ |
| 54 | @name检测依赖是否正常 |
| 55 | """ |
| 56 | try: |
| 57 | import psycopg2 |
| 58 | from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT |
| 59 | except: |
| 60 | os.system('btpip install psycopg2-binary') |
| 61 | try: |
| 62 | import psycopg2 |
| 63 | from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT |
| 64 | except: |
| 65 | return False |
| 66 | return True |
| 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): |
| 90 | """ |
no test coverage detected