Add system notice interface (#487)

Author: dividduang

backend/app/admin/api/v1/sys/__init__.py

@@ -10,6 +10,7 @@
 from backend.app.admin.api.v1.sys.dict_data import router as dict_data_router
 from backend.app.admin.api.v1.sys.dict_type import router as dict_type_router
 from backend.app.admin.api.v1.sys.menu import router as menu_router
+from backend.app.admin.api.v1.sys.notice import router as notice_router
 from backend.app.admin.api.v1.sys.role import router as role_router
 from backend.app.admin.api.v1.sys.user import router as user_router
 
@@ -25,3 +26,4 @@
 router.include_router(role_router, prefix='/roles', tags=['系统角色'])
 router.include_router(user_router, prefix='/users', tags=['系统用户'])
 router.include_router(data_rule_router, prefix='/data-rules', tags=['系统数据权限规则'])
+router.include_router(notice_router, prefix='/notices', tags=['系统通知公告'])

backend/app/admin/api/v1/sys/notice.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from backend.app.admin.schema.notice import CreateNoticeParam, GetNoticeListDetails, UpdateNoticeParam
+from backend.app.admin.service.notice_service import notice_service
+from backend.common.pagination import DependsPagination, paging_data
+from backend.common.response.response_schema import ResponseModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.common.security.permission import RequestPermission
+from backend.common.security.rbac import DependsRBAC
+from backend.database.db import CurrentSession
+
+router = APIRouter()
+
+
+@router.get('/{pk}', summary='获取通知公告详情', dependencies=[DependsJwtAuth])
+async def get_notice(pk: Annotated[int, Path(...)]) -> ResponseModel:
+    notice = await notice_service.get(pk=pk)
+    return response_base.success(data=notice)
+
+
+@router.get(
+    '',
+    summary='（模糊条件）分页获取所有通知公告',
+    dependencies=[
+        DependsJwtAuth,
+        DependsPagination,
+    ],
+)
+async def get_pagination_notice(db: CurrentSession) -> ResponseModel:
+    notice_select = await notice_service.get_select()
+    page_data = await paging_data(db, notice_select, GetNoticeListDetails)
+    return response_base.success(data=page_data)
+
+
+@router.post(
+    '',
+    summary='创建通知公告',
+    dependencies=[
+        Depends(RequestPermission('sys:notice:add')),
+        DependsRBAC,
+    ],
+)
+async def create_notice(obj: CreateNoticeParam) -> ResponseModel:
+    await notice_service.create(obj=obj)
+    return response_base.success()
+
+
+@router.put(
+    '/{pk}',
+    summary='更新通知公告',
+    dependencies=[
+        Depends(RequestPermission('sys:notice:edit')),
+        DependsRBAC,
+    ],
+)
+async def update_notice(pk: Annotated[int, Path(...)], obj: UpdateNoticeParam) -> ResponseModel:
+    count = await notice_service.update(pk=pk, obj=obj)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()
+
+
+@router.delete(
+    '',
+    summary='（批量）删除通知公告',
+    dependencies=[
+        Depends(RequestPermission('sys:notice:del')),
+        DependsRBAC,
+    ],
+)
+async def delete_notice(pk: Annotated[list[int], Query(...)]) -> ResponseModel:
+    count = await notice_service.delete(pk=pk)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()

backend/app/admin/crud/crud_notice.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from typing import Sequence
+
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy_crud_plus import CRUDPlus
+
+from backend.app.admin.model import Notice
+from backend.app.admin.schema.notice import CreateNoticeParam, UpdateNoticeParam
+
+
+class CRUDNotice(CRUDPlus[Notice]):
+    async def get(self, db: AsyncSession, pk: int) -> Notice | None:
+        """
+        获取系统通知公告
+
+        :param db:
+        :param pk:
+        :return:
+        """
+        return await self.select_model(db, pk)
+
+    async def get_list(self) -> Select:
+        """
+        获取系统通知公告列表
+
+        :return:
+        """
+        return await self.select_order('created_time', 'desc')
+
+    async def get_all(self, db: AsyncSession) -> Sequence[Notice]:
+        """
+        获取所有系统通知公告
+
+        :param db:
+        :return:
+        """
+        return await self.select_models(db)
+
+    async def create(self, db: AsyncSession, obj_in: CreateNoticeParam) -> None:
+        """
+        创建系统通知公告
+
+        :param db:
+        :param obj_in:
+        :return:
+        """
+        await self.create_model(db, obj_in)
+
+    async def update(self, db: AsyncSession, pk: int, obj_in: UpdateNoticeParam) -> int:
+        """
+        更新系统通知公告
+
+        :param db:
+        :param pk:
+        :param obj_in:
+        :return:
+        """
+        return await self.update_model(db, pk, obj_in)
+
+    async def delete(self, db: AsyncSession, pk: list[int]) -> int:
+        """
+        删除系统通知公告
+
+        :param db:
+        :param pk:
+        :return:
+        """
+        return await self.delete_model_by_column(db, allow_multiple=True, id__in=pk)
+
+
+notice_dao: CRUDNotice = CRUDNotice(Notice)

backend/app/admin/model/__init__.py

@@ -9,6 +9,7 @@
 from backend.app.admin.model.dict_type import DictType
 from backend.app.admin.model.login_log import LoginLog
 from backend.app.admin.model.menu import Menu
+from backend.app.admin.model.notice import Notice
 from backend.app.admin.model.opera_log import OperaLog
 from backend.app.admin.model.role import Role
 from backend.app.admin.model.user import User

backend/app/admin/model/notice.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from sqlalchemy import TEXT, String
+from sqlalchemy.dialects.mysql import LONGTEXT
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.common.model import Base, id_key
+
+
+class Notice(Base):
+    """系统通知公告"""
+
+    __tablename__ = 'sys_notice'
+
+    id: Mapped[id_key] = mapped_column(init=False)
+    title: Mapped[str] = mapped_column(String(50), comment='标题')
+    type: Mapped[int] = mapped_column(comment='类型（0：通知、1：公告）')
+    author: Mapped[str] = mapped_column(String(16), comment='作者')
+    source: Mapped[str] = mapped_column(String(50), comment='信息来源')
+    status: Mapped[int] = mapped_column(comment='状态（0：隐藏、1：显示）')
+    content: Mapped[str] = mapped_column(LONGTEXT().with_variant(TEXT, 'postgresql'), comment='内容')

backend/app/admin/schema/notice.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from datetime import datetime
+
+from pydantic import ConfigDict, Field
+
+from backend.common.enums import StatusType
+from backend.common.schema import SchemaBase
+
+
+class NoticeSchemaBase(SchemaBase):
+    title: str
+    type: int
+    author: str
+    source: str
+    status: StatusType = Field(StatusType.enable)
+    content: str
+
+
+class CreateNoticeParam(NoticeSchemaBase):
+    pass
+
+
+class UpdateNoticeParam(NoticeSchemaBase):
+    pass
+
+
+class GetNoticeListDetails(NoticeSchemaBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    created_time: datetime
+    updated_time: datetime | None = None

backend/app/admin/service/notice_service.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from typing import Sequence
+
+from sqlalchemy import Select
+
+from backend.app.admin.crud.crud_notice import notice_dao
+from backend.app.admin.model import Notice
+from backend.app.admin.schema.notice import CreateNoticeParam, UpdateNoticeParam
+from backend.common.exception import errors
+from backend.database.db import async_db_session
+
+
+class NoticeService:
+    @staticmethod
+    async def get(*, pk: int) -> Notice:
+        async with async_db_session() as db:
+            notice = await notice_dao.get(db, pk)
+            if not notice:
+                raise errors.NotFoundError(msg='通知公告不存在')
+            return notice
+
+    @staticmethod
+    async def get_select() -> Select:
+        return await notice_dao.get_list()
+
+    @staticmethod
+    async def get_all() -> Sequence[Notice]:
+        async with async_db_session() as db:
+            notices = await notice_dao.get_all(db)
+            return notices
+
+    @staticmethod
+    async def create(*, obj: CreateNoticeParam) -> None:
+        async with async_db_session.begin() as db:
+            await notice_dao.create(db, obj)
+
+    @staticmethod
+    async def update(*, pk: int, obj: UpdateNoticeParam) -> int:
+        async with async_db_session.begin() as db:
+            notice = await notice_dao.get(db, pk)
+            if not notice:
+                raise errors.NotFoundError(msg='通知公告不存在')
+            count = await notice_dao.update(db, pk, obj)
+            return count
+
+    @staticmethod
+    async def delete(*, pk: list[int]) -> int:
+        async with async_db_session.begin() as db:
+            count = await notice_dao.delete(db, pk)
+            return count
+
+
+notice_service: NoticeService = NoticeService()