Fix some bugs

Author: wu-clan

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

@@ -17,13 +17,15 @@
 
 
 @router.get('/all', summary='获取所有任务调度', dependencies=[DependsJwtAuth])
-async def get_all_task_schedulers() -> ResponseModel:
+async def get_all_task_schedulers() -> ResponseSchemaModel[list[GetTaskSchedulerDetail]]:
     schedulers = await task_scheduler_service.get_all()
     return response_base.success(data=schedulers)
 
 
 @router.get('/{pk}', summary='获取任务调度详情', dependencies=[DependsJwtAuth])
-async def get_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
+async def get_task_scheduler(
+    pk: Annotated[int, Path(description='任务调度 ID')],
+) -> ResponseSchemaModel[GetTaskSchedulerDetail]:
     task_scheduler = await task_scheduler_service.get(pk=pk)
     return response_base.success(data=task_scheduler)
 
@@ -54,7 +56,7 @@ async def get_task_scheduler_paged(
         DependsRBAC,
     ],
 )
-async def create_task_scheduler(obj: CreateTaskSchedulerParam):
+async def create_task_scheduler(obj: CreateTaskSchedulerParam) -> ResponseModel:
     await task_scheduler_service.create(obj=obj)
     return response_base.success()
 
@@ -67,7 +69,9 @@ async def create_task_scheduler(obj: CreateTaskSchedulerParam):
         DependsRBAC,
     ],
 )
-async def update_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')], obj: UpdateTaskSchedulerParam):
+async def update_task_scheduler(
+    pk: Annotated[int, Path(description='任务调度 ID')], obj: UpdateTaskSchedulerParam
+) -> ResponseModel:
     count = await task_scheduler_service.update(pk=pk, obj=obj)
     if count > 0:
         return response_base.success()
@@ -82,7 +86,7 @@ async def update_task_scheduler(pk: Annotated[int, Path(description='任务调
         DependsRBAC,
     ],
 )
-async def update_task_scheduler_status(pk: Annotated[int, Path(description='任务调度 ID')]):
+async def update_task_scheduler_status(pk: Annotated[int, Path(description='任务调度 ID')]) -> ResponseModel:
     count = await task_scheduler_service.update_status(pk=pk)
     if count > 0:
         return response_base.success()
@@ -97,7 +101,7 @@ async def update_task_scheduler_status(pk: Annotated[int, Path(description='任
         DependsRBAC,
     ],
 )
-async def delete_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]):
+async def delete_task_scheduler(pk: Annotated[int, Path(description='任务调度 ID')]) -> ResponseModel:
     count = await task_scheduler_service.delete(pk=pk)
     if count > 0:
         return response_base.success()
@@ -112,7 +116,7 @@ async def delete_task_scheduler(pk: Annotated[int, Path(description='任务调
         DependsRBAC,
     ],
 )
-async def execute_task(pk: Annotated[int, Path(description='任务调度 ID')]) -> ResponseSchemaModel[str]:
+async def execute_task(pk: Annotated[int, Path(description='任务调度 ID')]) -> ResponseModel:
     await task_scheduler_service.execute(pk=pk)
     return response_base.success()
 

backend/app/task/celery.py

@@ -43,15 +43,13 @@ def init_celery() -> celery.Celery:
             'group': OVERWRITE_CELERY_RESULT_GROUP_TABLE_NAME,
         },
         result_extended=True,
-        # result_expires=0,  # 任务结果自动清理
+        # result_expires=0,  # 任务结果自动清理，0 或 None 表示不清理
         beat_schedule=LOCAL_BEAT_SCHEDULE,
         beat_scheduler='app.task.utils.schedulers:DatabaseScheduler',
         task_cls='app.task.tasks.base:TaskBase',
         task_track_started=True,
         enable_utc=False,
         timezone=settings.DATETIME_TIMEZONE,
-        # TODO: Update this work if celery version >= 6.0.0
-        worker_pool=celery_aio_pool.pool.AsyncIOPool,
     )
 
     # 自动发现任务

backend/app/task/crud/crud_scheduler.py

@@ -89,7 +89,7 @@ async def set_status(self, db: AsyncSession, pk: int, status: bool) -> int:
         :param status: 状态
         :return:
         """
-        return await self.update_model(db, pk, {'status': status})
+        return await self.update_model(db, pk, {'enabled': status})
 
     async def delete(self, db: AsyncSession, pk: int) -> int:
         """

backend/app/task/model/result.py

@@ -5,4 +5,5 @@
 OVERWRITE_CELERY_RESULT_TABLE_NAME = 'task_result'
 OVERWRITE_CELERY_RESULT_GROUP_TABLE_NAME = 'task_group_result'
 
+# 重写表名配置
 TaskResult.configure(name=OVERWRITE_CELERY_RESULT_TABLE_NAME)

backend/app/task/model/scheduler.py

@@ -9,10 +9,10 @@
     Boolean,
     DateTime,
     String,
+    event,
 )
 from sqlalchemy.dialects.mysql import LONGTEXT
 from sqlalchemy.dialects.postgresql import INTEGER, TEXT
-from sqlalchemy.event import listen
 from sqlalchemy.orm import Mapped, mapped_column
 
 from backend.common.exception import errors
@@ -66,26 +66,28 @@ class TaskScheduler(Base):
 
     @staticmethod
     def before_insert_or_update(mapper, connection, target):
-        """插入或更新前的验证"""
+        print('before_insert_or_update', mapper, connection, target)
         if target.expire_seconds is not None and target.expire_time:
             raise errors.ConflictError(msg='expires 和 expire_seconds 只能设置一个')
 
     @classmethod
     def changed(cls, mapper, connection, target):
+        print('changed', mapper, connection, target)
         if not target.no_changes:
             cls.update_changed(mapper, connection, target)
 
     @classmethod
     def update_changed(cls, mapper, connection, target):
+        print('update_changed', mapper, connection, target)
         now = timezone.now()
-        last_update = asyncio.create_task(redis_client.get(f'{settings.CELERY_REDIS_PREFIX}last_update'))
+        last_update = asyncio.create_task(redis_client.get(f'{settings.CELERY_REDIS_PREFIX}:last_update'))
         if not last_update:
-            asyncio.create_task(redis_client.set(f'{settings.CELERY_REDIS_PREFIX}last_update', timezone.to_str(now)))
+            asyncio.create_task(redis_client.set(f'{settings.CELERY_REDIS_PREFIX}:last_update', timezone.to_str(now)))
 
 
 # 事件监听器
-listen(TaskScheduler, 'before_insert', TaskScheduler.before_insert_or_update)
-listen(TaskScheduler, 'before_update', TaskScheduler.before_insert_or_update)
-listen(TaskScheduler, 'after_insert', TaskScheduler.update_changed)
-listen(TaskScheduler, 'after_delete', TaskScheduler.update_changed)
-listen(TaskScheduler, 'after_update', TaskScheduler.changed)
+event.listen(TaskScheduler, 'before_insert', TaskScheduler.before_insert_or_update)
+event.listen(TaskScheduler, 'before_update', TaskScheduler.before_insert_or_update)
+event.listen(TaskScheduler, 'after_insert', TaskScheduler.update_changed)
+event.listen(TaskScheduler, 'after_delete', TaskScheduler.update_changed)
+event.listen(TaskScheduler, 'after_update', TaskScheduler.changed)

backend/app/task/schema/scheduler.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 
 from pydantic import ConfigDict, Field
+from pydantic.types import JsonValue
 
 from backend.app.task.enums import PeriodType, TaskSchedulerType
 from backend.common.schema import SchemaBase
@@ -13,8 +14,8 @@ class TaskSchedulerSchemeBase(SchemaBase):
 
     name: str = Field(description='任务名称')
     task: str = Field(description='要运行的 Celery 任务（模块化字符串）')
-    args: str | None = Field(default='[]', description='任务可接收的位置参数')
-    kwargs: str | None = Field(default='{}', description='任务可接收的关键字参数')
+    args: JsonValue | None = Field(default='[]', description='任务可接收的位置参数')
+    kwargs: JsonValue | None = Field(default='{}', description='任务可接收的关键字参数')
     queue: str | None = Field(default=None, description='CELERY_TASK_QUEUES 中定义的队列')
     exchange: str | None = Field(default=None, description='低级别 AMQP 路由的交换机')
     routing_key: str | None = Field(default=None, description='低级别 AMQP 路由的路由密钥')
@@ -24,7 +25,7 @@ class TaskSchedulerSchemeBase(SchemaBase):
     last_run_time: datetime | None = Field(default=None, description='任务最后触发的时间')
     type: TaskSchedulerType = Field(default=TaskSchedulerType.INTERVAL, description='任务调度类型（0间隔 1定时）')
     interval_every: int | None = Field(default=None, description='任务再次运行前的间隔周期数')
-    interval_period: PeriodType = Field(default=None, description='任务运行之间的周期类型')
+    interval_period: PeriodType | 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='运行的星期，"*" 表示全部')
@@ -47,3 +48,7 @@ class GetTaskSchedulerDetail(TaskSchedulerSchemeBase):
     """任务调度详情"""
 
     model_config = ConfigDict(from_attributes=True)
+
+    id: int = Field(description='任务调度 ID')
+    created_time: datetime = Field(description='创建时间')
+    updated_time: datetime | None = Field(None, description='更新时间')

backend/app/task/service/scheduler_service.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+import json
+
 from typing import Sequence
 
 from sqlalchemy import Select
@@ -125,19 +127,6 @@ async def delete(*, pk) -> int:
             count = await task_scheduler_dao.delete(db, pk)
             return count
 
-    @staticmethod
-    async def revoke(*, task_id: str) -> None:
-        """
-        撤销指定的任务
-
-        :param task_id: 任务 UUID
-        :return:
-        """
-        workers = await run_in_threadpool(celery_app.control.ping())
-        if not workers:
-            raise errors.ServerError(msg='Celery Worker 未启动，请联系超级管理员')
-        celery_app.control.revoke(task_id, terminate=True)
-
     @staticmethod
     async def execute(*, pk: int) -> None:
         """
@@ -147,17 +136,30 @@ async def execute(*, pk: int) -> None:
         :return:
         """
         async with async_db_session() as db:
-            workers = await run_in_threadpool(celery_app.control.ping())
+            workers = await run_in_threadpool(celery_app.control.ping, timeout=0.5)
             if not workers:
-                raise errors.ServerError(msg='Celery Worker 未启动，请联系超级管理员')
+                raise errors.ServerError(msg='Celery Worker 暂不可用，请稍后重试')
             task_scheduler = await task_scheduler_dao.get(db, pk)
             if not task_scheduler:
                 raise errors.NotFoundError(msg='任务调度不存在')
             celery_app.send_task(
                 name=task_scheduler.task,
-                args=task_scheduler.args,
-                kwargs=task_scheduler.kwargs,
+                args=json.loads(task_scheduler.args),
+                kwargs=json.loads(task_scheduler.kwargs),
             )
 
+    @staticmethod
+    async def revoke(*, task_id: str) -> None:
+        """
+        撤销指定的任务
+
+        :param task_id: 任务 UUID
+        :return:
+        """
+        workers = await run_in_threadpool(celery_app.control.ping, timeout=0.5)
+        if not workers:
+            raise errors.ServerError(msg='Celery Worker 暂不可用，请稍后重试')
+        celery_app.control.revoke(task_id)
+
 
 task_scheduler_service: TaskSchedulerService = TaskSchedulerService()

backend/app/task/utils/schedulers.py

@@ -309,14 +309,15 @@ async def get_all_task_schedules(self):
     def schedule_changed(self) -> bool:
         """任务调度变更状态"""
         now = timezone.now()
-        last_update = run_await(redis_client.get)(f'{settings.CELERY_REDIS_PREFIX}last_update')
+        last_update = run_await(redis_client.get)(f'{settings.CELERY_REDIS_PREFIX}:last_update')
         if not last_update:
-            run_await(redis_client.set)(f'{settings.CELERY_REDIS_PREFIX}last_update', timezone.to_str(now))
+            run_await(redis_client.set)(f'{settings.CELERY_REDIS_PREFIX}:last_update', timezone.to_str(now))
             return False
+
+        last, ts = self._last_update, timezone.from_str(last_update)
         try:
-            if last_update and self._last_update:
-                if timezone.from_str(last_update) > self._last_update:
-                    return True
+            if ts and ts > (last if last else ts):
+                return True
         finally:
             self._last_update = now
 
@@ -337,10 +338,10 @@ def sync(self):
                 try:
                     tasks = self.schedule
                     run_await(tasks[name].save)()
-                    logger.debug(f'保存任务 {name} 状态到数据库')
+                    logger.debug(f'保存任务 {name} 最新状态到数据库')
                     _tried.add(name)
                 except KeyError as e:
-                    logger.error(f'保存任务 {name} 状态失败：{e} ')
+                    logger.error(f'保存任务 {name} 最新状态失败：{e} ')
                     _failed.add(name)
         except DatabaseError as e:
             logger.exception('同步时出现数据库错误: %r', e)

backend/app/task/utils/tzcrontab.py

@@ -83,4 +83,4 @@ def crontab_verify(filed: Literal['m', 'h', 'dow', 'dom', 'moy'], value: str, ra
         if raise_exc:
             raise errors.RequestError(msg=f'crontab 值 {value} 非法')
 
-    raise valid
+    return valid

backend/core/conf.py

@@ -209,7 +209,7 @@ class Settings(BaseSettings):
 
     # 基础配置
     CELERY_BROKER: Literal['rabbitmq', 'redis'] = 'redis'
-    CELERY_REDIS_PREFIX: str = 'fba:celery:'
+    CELERY_REDIS_PREFIX: str = 'fba:celery'
     CELERY_TASK_MAX_RETRIES: int = 5
 
     # Plugin Code Generator