@取pgsql数据库对像 By sid @sid 数据库分类,0:本地
(self, sid: Union[int, str] = 0, conn_config=None)
| 291 | return base_list |
| 292 | |
| 293 | def get_sql_obj_by_sid(self, sid: Union[int, str] = 0, conn_config=None): |
| 294 | """ |
| 295 | @取pgsql数据库对像 By sid |
| 296 | @sid 数据库分类,0:本地 |
| 297 | """ |
| 298 | if isinstance(sid, str): |
| 299 | sid = int(sid) |
| 300 | |
| 301 | if sid: |
| 302 | if not conn_config: conn_config = public.M('database_servers').where("id=? AND LOWER(db_type)=LOWER('pgsql')", sid).find() |
| 303 | db_obj = panelPgsql() |
| 304 | |
| 305 | try: |
| 306 | db_obj = db_obj.set_host(host=conn_config["db_host"], port=conn_config["db_port"], database=None, user=conn_config["db_user"], password=conn_config["db_password"]) |
| 307 | except Exception as e: |
| 308 | raise public.PanelError(e) |
| 309 | else: |
| 310 | db_obj = panelPgsql() |
| 311 | return db_obj |
| 312 | |
| 313 | def get_sql_obj(self, db_name): |
| 314 | """ |
no test coverage detected