refactor(module): simplify service instantiation with generics

This commit is contained in:
InfinityPacer
2024-09-27 04:04:56 +08:00
parent 5efcd6e6be
commit 2ce3ddb75a
19 changed files with 244 additions and 242 deletions
+1 -19
View File
@@ -1,23 +1,5 @@
from typing import List
from app.db.systemconfig_oper import SystemConfigOper
from app.schemas.system import DownloaderConf
from app.schemas.types import SystemConfigKey
class DownloaderHelper:
"""
下载器帮助类
"""
def __init__(self):
self.systemconfig = SystemConfigOper()
def get_downloaders(self) -> List[DownloaderConf]:
"""
获取下载器
"""
downloader_confs: List[dict] = self.systemconfig.get(SystemConfigKey.Downloaders)
if not downloader_confs:
return []
return [DownloaderConf(**conf) for conf in downloader_confs]
pass
+1 -19
View File
@@ -1,23 +1,5 @@
from typing import List
from app.db.systemconfig_oper import SystemConfigOper
from app.schemas import MediaServerConf
from app.schemas.types import SystemConfigKey
class MediaServerHelper:
"""
媒体服务器帮助类
"""
def __init__(self):
self.systemconfig = SystemConfigOper()
def get_mediaservers(self) -> List[MediaServerConf]:
"""
获取媒体服务器
"""
mediaserver_confs: List[dict] = self.systemconfig.get(SystemConfigKey.MediaServers)
if not mediaserver_confs:
return []
return [MediaServerConf(**conf) for conf in mediaserver_confs]
pass
+1 -38
View File
@@ -1,42 +1,5 @@
from typing import List, Optional
from app.db.systemconfig_oper import SystemConfigOper
from app.schemas import NotificationConf, NotificationSwitchConf
from app.schemas.types import SystemConfigKey, NotificationType
class NotificationHelper:
"""
消息通知渠道帮助类
"""
def __init__(self):
self.systemconfig = SystemConfigOper()
def get_clients(self) -> List[NotificationConf]:
"""
获取消息通知渠道
"""
client_confs: List[dict] = self.systemconfig.get(SystemConfigKey.Notifications)
if not client_confs:
return []
return [NotificationConf(**conf) for conf in client_confs]
def get_switchs(self) -> List[NotificationSwitchConf]:
"""
获取消息通知场景开关
"""
switchs: List[dict] = self.systemconfig.get(SystemConfigKey.NotificationSwitchs)
if not switchs:
return []
return [NotificationSwitchConf(**switch) for switch in switchs]
def get_switch(self, mtype: NotificationType) -> Optional[str]:
"""
获取消息通知场景开关
"""
switchs = self.get_switchs()
for switch in switchs:
if switch.type == mtype.value:
return switch.action
return None
pass
+65
View File
@@ -0,0 +1,65 @@
from typing import List, Type, Optional
from app.db.systemconfig_oper import SystemConfigOper
from app.schemas import DownloaderConf, MediaServerConf, NotificationConf, NotificationSwitchConf
from app.schemas.types import SystemConfigKey, NotificationType
class ServiceConfigHelper:
"""
配置帮助类,获取不同类型的服务配置
"""
@staticmethod
def get_configs(config_key: SystemConfigKey, conf_type: Type) -> List:
"""
通用获取配置的方法,根据 config_key 获取相应的配置并返回指定类型的配置列表
:param config_key: 系统配置的 key
:param conf_type: 用于实例化配置对象的类类型
:return: 配置对象列表
"""
config_data = SystemConfigOper().get(config_key)
if not config_data:
return []
# 直接使用 conf_type 来实例化配置对象
return [conf_type(**conf) for conf in config_data]
@staticmethod
def get_downloader_configs() -> List[DownloaderConf]:
"""
获取下载器的配置
"""
return ServiceConfigHelper.get_configs(SystemConfigKey.Downloaders, DownloaderConf)
@staticmethod
def get_mediaserver_configs() -> List[MediaServerConf]:
"""
获取媒体服务器的配置
"""
return ServiceConfigHelper.get_configs(SystemConfigKey.MediaServers, MediaServerConf)
@staticmethod
def get_notification_configs() -> List[NotificationConf]:
"""
获取消息通知渠道的配置
"""
return ServiceConfigHelper.get_configs(SystemConfigKey.Notifications, NotificationConf)
@staticmethod
def get_notification_switches() -> List[NotificationSwitchConf]:
"""
获取消息通知场景的开关
"""
return ServiceConfigHelper.get_configs(SystemConfigKey.NotificationSwitchs, NotificationSwitchConf)
@staticmethod
def get_notification_switch(mtype: NotificationType) -> Optional[str]:
"""
获取消息通知场景开关
"""
switchs = ServiceConfigHelper.get_notification_switches()
for switch in switchs:
if switch.type == mtype.value:
return switch.action
return None