add 消息类型&API

app/api/endpoints/message.py

@@ -1,14 +1,20 @@
-from typing import Union, Any
+from typing import Union, Any, List
 
-from fastapi import APIRouter, BackgroundTasks
+from fastapi import APIRouter, BackgroundTasks, Depends
 from fastapi import Request
+from sqlalchemy.orm import Session
 from starlette.responses import PlainTextResponse
 
 from app import schemas
 from app.chain.message import MessageChain
 from app.core.config import settings
+from app.core.security import verify_token
+from app.db import get_db
+from app.db.systemconfig_oper import SystemConfigOper
 from app.log import logger
 from app.modules.wechat.WXBizMsgCrypt3 import WXBizMsgCrypt
+from app.schemas import Notification
+from app.schemas.types import SystemConfigKey, NotificationType
 
 router = APIRouter()
 
@@ -54,3 +60,35 @@ def wechat_verify(echostr: str, msg_signature: str,
         logger.error("微信请求验证失败 VerifyURL ret: %s" % str(ret))
     # 验证URL成功，将sEchoStr返回给企业号
     return PlainTextResponse(sEchoStr)
+
+
+@router.get("/switchs", summary="查询通知消息渠道开关", response_model=List[Notification])
+def read_switchs(db: Session = Depends(get_db),
+                 _: schemas.TokenPayload = Depends(verify_token)) -> Any:
+    """
+    查询通知消息渠道开关
+    """
+    return_list = []
+    # 读取数据库
+    switchs = SystemConfigOper(db).get(SystemConfigKey.NotificationChannels)
+    if not switchs:
+        for noti in NotificationType:
+            return_list.append(Notification(mtype=noti.value, switch=True))
+
+    return return_list
+
+
+@router.put("/switchs", summary="设置通知消息渠道开关", response_model=schemas.Response)
+def set_switchs(switchs: List[Notification],
+                db: Session = Depends(get_db),
+                _: schemas.TokenPayload = Depends(verify_token)) -> Any:
+    """
+    查询通知消息渠道开关
+    """
+    switch_list = []
+    for switch in switchs:
+        switch_list.append(switch.dict())
+    # 存入数据库
+    SystemConfigOper(db).set(SystemConfigKey.NotificationChannels, switch_list)
+
+    return schemas.Response(success=True)