mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-30 04:27:40 +08:00
58 lines
1.3 KiB
Python
58 lines
1.3 KiB
Python
from typing import Tuple, Union
|
|
|
|
from app.runtime.config import settings
|
|
from app.application.database import get_configured_database_health
|
|
from app.modules import _ModuleBase
|
|
from app.schemas.types import ModuleType, OtherModulesType
|
|
|
|
|
|
class PostgreSQLModule(_ModuleBase):
|
|
"""
|
|
PostgreSQL 数据库模块
|
|
"""
|
|
|
|
def init_module(self) -> None:
|
|
pass
|
|
|
|
@staticmethod
|
|
def get_name() -> str:
|
|
return "PostgreSQL"
|
|
|
|
@staticmethod
|
|
def get_type() -> ModuleType:
|
|
"""
|
|
获取模块类型
|
|
"""
|
|
return ModuleType.Other
|
|
|
|
@staticmethod
|
|
def get_subtype() -> OtherModulesType:
|
|
"""
|
|
获取模块子类型
|
|
"""
|
|
return OtherModulesType.PostgreSQL
|
|
|
|
@staticmethod
|
|
def get_priority() -> int:
|
|
"""
|
|
获取模块优先级,数字越小优先级越高,只有同一接口下优先级才生效
|
|
"""
|
|
return 0
|
|
|
|
def init_setting(self) -> Tuple[str, Union[str, bool]]:
|
|
pass
|
|
|
|
def stop(self) -> None:
|
|
pass
|
|
|
|
def test(self):
|
|
"""
|
|
测试模块连接性
|
|
"""
|
|
if settings.DB_TYPE != "postgresql":
|
|
return None
|
|
error = get_configured_database_health().test()
|
|
if error:
|
|
return False, f"PostgreSQL连接失败:{error}"
|
|
return True, ""
|