Refine the business codes

Author: wu-clan

backend/app/task/api/v1/scheduler.py

@@ -2,26 +2,30 @@
 # -*- coding: utf-8 -*-
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Path
+from fastapi import APIRouter, Depends, Path, Query
 
+from backend.app.task.schema.scheduler import CreateTaskSchedulerParam, GetTaskSchedulerDetail, UpdateTaskSchedulerParam
 from backend.app.task.service.scheduler_service import task_scheduler_service
-from backend.common.pagination import DependsPagination
+from backend.common.pagination import DependsPagination, PageData, paging_data
 from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, 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('/all', summary='获取所有任务调度', dependencies=[DependsJwtAuth])
-async def get_all_schedulers() -> ResponseModel:
-    return response_base.success()
+async def get_all_task_schedulers() -> ResponseModel:
+    schedulers = await task_scheduler_service.get_all()
+    return response_base.success(data=schedulers)
 
 
 @router.get('/{pk}', summary='获取任务调度详情', dependencies=[DependsJwtAuth])
-async def get_scheduler():
-    return response_base.success()
+async def get_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
+    task_scheduler = await task_scheduler_service.get(pk=pk)
+    return response_base.success(data=task_scheduler)
 
 
 @router.get(
@@ -32,8 +36,14 @@ async def get_scheduler():
         DependsPagination,
     ],
 )
-async def get_scheduler_paged():
-    return response_base.success()
+async def get_task_scheduler_paged(
+    db: CurrentSession,
+    name: Annotated[int, Path(description='任务调度名称')] = None,
+    type: Annotated[int | None, Query(description='任务调度类型')] = None,
+) -> ResponseSchemaModel[PageData[GetTaskSchedulerDetail]]:
+    task_scheduler_select = await task_scheduler_service.get_select(name=name, type=type)
+    page_data = await paging_data(db, task_scheduler_select)
+    return response_base.success(data=page_data)
 
 
 @router.post(
@@ -44,7 +54,8 @@ async def get_scheduler_paged():
         DependsRBAC,
     ],
 )
-async def create_scheduler():
+async def create_task_scheduler(obj: CreateTaskSchedulerParam):
+    await task_scheduler_service.create(obj=obj)
     return response_base.success()
 
 
@@ -56,8 +67,11 @@ async def create_scheduler():
         DependsRBAC,
     ],
 )
-async def update_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
-    return response_base.success()
+async def update_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')], obj: UpdateTaskSchedulerParam):
+    count = await task_scheduler_service.update(pk=pk, obj=obj)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()
 
 
 @router.put(
@@ -68,8 +82,11 @@ async def update_scheduler(pk: Annotated[int, Path(description='任务调度 ID'
         DependsRBAC,
     ],
 )
-async def update_scheduler_status(pk: Annotated[int, Path(description='任务调度 ID')]):
-    return response_base.success()
+async def update_task_scheduler_status(pk: Annotated[int, Path(description='任务调度 ID')]):
+    count = await task_scheduler_service.update_status(pk=pk)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()
 
 
 @router.delete(
@@ -80,8 +97,11 @@ async def update_scheduler_status(pk: Annotated[int, Path(description='任务调
         DependsRBAC,
     ],
 )
-async def delete_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
-    return response_base.success()
+async def delete_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
+    count = await task_scheduler_service.delete(pk=pk)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()
 
 
 @router.post(
@@ -93,8 +113,8 @@ async def delete_scheduler(pk: Annotated[int, Path(description='任务调度 ID'
     ],
 )
 async def execute_task(pk: Annotated[int, Path(description='任务调度 ID')]) -> ResponseSchemaModel[str]:
-    task = task_scheduler_service.execute(pk=pk)
-    return response_base.success(data=task)
+    await task_scheduler_service.execute(pk=pk)
+    return response_base.success()
 
 
 @router.delete(
@@ -106,5 +126,5 @@ async def execute_task(pk: Annotated[int, Path(description='任务调度 ID')])
     ],
 )
 async def revoke_task(task_id: Annotated[str, Path(description='任务 UUID')]) -> ResponseModel:
-    task_scheduler_service.revoke(task_id=task_id)
+    await task_scheduler_service.revoke(task_id=task_id)
     return response_base.success()

backend/app/task/crud/crud_scheduler.py

@@ -0,0 +1,105 @@
+#!/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.task.model import TaskScheduler
+from backend.app.task.schema.scheduler import CreateTaskSchedulerParam, UpdateTaskSchedulerParam
+
+
+class CRUDTaskScheduler(CRUDPlus[TaskScheduler]):
+    """任务调度数据库操作类"""
+
+    async def get(self, db: AsyncSession, pk: int) -> TaskScheduler | None:
+        """
+        获取任务调度
+
+        :param db: 数据库会话
+        :param pk: 任务调度 ID
+        :return:
+        """
+        return await task_scheduler_dao.select_model(db, pk)
+
+    async def get_all(self, db: AsyncSession) -> Sequence[TaskScheduler]:
+        """
+        获取所有任务调度
+
+        :param db: 数据库会话
+        :return:
+        """
+        return await self.select_models(db)
+
+    async def get_list(self, name: str | None, type: int | None) -> Select:
+        """
+        获取任务调度列表
+
+        :param name: 任务调度名称
+        :param type: 任务调度类型
+        :return:
+        """
+        filters = {}
+
+        if name is not None:
+            filters['name__like'] = f'%{name}%'
+        if type is not None:
+            filters['type'] = type
+
+        return await self.select_order('id', **filters)
+
+    async def get_by_name(self, db: AsyncSession, name: str) -> TaskScheduler | None:
+        """
+        通过名称获取任务调度
+
+        :param db: 数据库会话
+        :param name: 任务调度名称
+        :return:
+        """
+        return await self.select_model_by_column(db, name=name)
+
+    async def create(self, db: AsyncSession, obj: CreateTaskSchedulerParam) -> None:
+        """
+        创建任务调度
+
+        :param db: 数据库会话
+        :param obj: 创建任务调度参数
+        :return:
+        """
+        await self.create_model(db, obj)
+
+    async def update(self, db: AsyncSession, pk: int, obj: UpdateTaskSchedulerParam) -> int:
+        """
+        更新任务调度
+
+        :param db: 数据库会话
+        :param pk: 任务调度 ID
+        :param obj: 更新任务调度参数
+        :return:
+        """
+        return await self.update_model(db, pk, obj)
+
+    async def set_status(self, db: AsyncSession, pk: int, status: bool) -> int:
+        """
+        设置任务调度状态
+
+        :param db: 数据库会话
+        :param pk: 任务调度 ID
+        :param status: 状态
+        :return:
+        """
+        return await self.update_model(db, pk, {'status': status})
+
+    async def delete(self, db: AsyncSession, pk: int) -> int:
+        """
+        删除任务调度
+
+        :param db: 数据库会话
+        :param pk: 任务调度 ID
+        :return:
+        """
+        return await self.delete_model(db, pk)
+
+
+task_scheduler_dao: CRUDTaskScheduler = CRUDTaskScheduler(TaskScheduler)

backend/app/task/schema/scheduler.py

@@ -2,13 +2,13 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime
 
-from pydantic import Field
+from pydantic import ConfigDict, Field
 
 from backend.app.task.enums import PeriodType, TaskSchedulerType
 from backend.common.schema import SchemaBase
 
 
-class TaskSchedulerParam(SchemaBase):
+class TaskSchedulerSchemeBase(SchemaBase):
     """任务调度参数"""
 
     name: str = Field(description='任务名称')
@@ -25,16 +25,25 @@ class TaskSchedulerParam(SchemaBase):
     type: TaskSchedulerType = Field(default=TaskSchedulerType.INTERVAL, description='任务调度类型（0间隔 1定时）')
     interval_every: int | None = Field(default=None, description='任务再次运行前的间隔周期数')
     interval_period: PeriodType = Field(default=None, description='任务运行之间的周期类型')
-    crontab_minute: str | None = Field(default=None, description='运行的分钟，"*" 表示全部')
-    crontab_hour: str | None = Field(default=None, description='运行的小时，"*" 表示全部')
-    crontab_day_of_week: str | None = Field(default=None, description='运行的星期，"*" 表示全部')
-    crontab_day_of_month: str | None = Field(default=None, description='运行的每月日期，"*" 表示全部')
-    crontab_month_of_year: str | None = Field(default=None, description='运行的月份，"*" 表示全部')
+    crontab_minute: str | None = Field(default='*', description='运行的分钟，"*" 表示全部')
+    crontab_hour: str | None = Field(default='*', description='运行的小时，"*" 表示全部')
+    crontab_day_of_week: str | None = Field(default='*', description='运行的星期，"*" 表示全部')
+    crontab_day_of_month: str | None = Field(default='*', description='运行的每月日期，"*" 表示全部')
+    crontab_month_of_year: str | None = Field(default='*', description='运行的月份，"*" 表示全部')
     one_off: bool = Field(default=False, description='是否仅运行一次')
-    enabled: bool = Field(default=True, description='是否启用任务')
     total_run_count: int = Field(default=0, description='任务触发的总次数')
     remark: str | None = Field(default=None, description='备注')
 
 
-class CreateTaskSchedulerParam(TaskSchedulerParam):
+class CreateTaskSchedulerParam(TaskSchedulerSchemeBase):
     """创建任务调度参数"""
+
+
+class UpdateTaskSchedulerParam(TaskSchedulerSchemeBase):
+    """更新任务调度参数"""
+
+
+class GetTaskSchedulerDetail(TaskSchedulerSchemeBase):
+    """任务调度详情"""
+
+    model_config = ConfigDict(from_attributes=True)