Add schedule task demo that contains params (#746)

Author: wu-clan

backend/app/task/crud/crud_result.py

@@ -35,7 +35,7 @@ async def get_list(self, name: str | None, task_id: str | None) -> Select:
         if task_id is not None:
             filters['task_id'] = task_id
 
-        return await self.select_order('id', **filters)
+        return await self.select_order('id', 'desc', **filters)
 
     async def delete(self, db: AsyncSession, pks: list[int]) -> int:
         """

backend/app/task/schema/result.py

@@ -3,8 +3,9 @@
 from datetime import datetime
 from typing import Any
 
-from pydantic import ConfigDict, Field
+from pydantic import ConfigDict, Field, field_serializer
 
+from backend.app.task import celery_app
 from backend.common.schema import SchemaBase
 
 
@@ -36,3 +37,7 @@ class GetTaskResultDetail(TaskResultSchemaBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int = Field(description='任务结果 ID')
+
+    @field_serializer('args', 'kwargs', when_used='unless-none')
+    def serialize_params(self, value: bytes | None, _info) -> Any:
+        return celery_app.backend.decode(value)

backend/app/task/tasks/beat.py

@@ -4,15 +4,22 @@
 
 from backend.app.task.utils.tzcrontab import TzAwareCrontab
 
+# 参考：https://docs.celeryq.dev/en/stable/userguide/periodic-tasks.html
 LOCAL_BEAT_SCHEDULE = {
     '测试同步任务': {
         'task': 'task_demo',
-        'schedule': schedule(5),
+        'schedule': schedule(30),
     },
     '测试异步任务': {
         'task': 'task_demo_async',
         'schedule': TzAwareCrontab('1'),
     },
+    '测试传参任务': {
+        'task': 'task_demo_params',
+        'schedule': TzAwareCrontab('1'),
+        'args': ['你好，'],
+        'kwargs': {'b': '世界'},
+    },
     '清理操作日志': {
         'task': 'backend.app.task.tasks.db_log.tasks.delete_db_opera_log',
         'schedule': TzAwareCrontab('0', '0', day_of_week='6'),

backend/app/task/tasks/db_log/tasks.py

@@ -1,18 +1,19 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from celery import shared_task
+
 from backend.app.admin.service.login_log_service import login_log_service
 from backend.app.admin.service.opera_log_service import opera_log_service
-from backend.app.task.celery import celery_app
 
 
-@celery_app.task
+@shared_task
 async def delete_db_opera_log() -> str:
     """自动删除数据库操作日志"""
     await opera_log_service.delete_all()
     return 'Success'
 
 
-@celery_app.task
+@shared_task
 async def delete_db_login_log() -> str:
     """自动删除数据库登录日志"""
     await login_log_service.delete_all()

backend/app/task/tasks/tasks.py

@@ -10,12 +10,18 @@
 @celery_app.task(name='task_demo')
 def task_demo() -> str:
     """示例任务，模拟耗时操作"""
-    sleep(20)
+    sleep(30)
     return 'test async'
 
 
 @celery_app.task(name='task_demo_async')
 async def task_demo_async() -> str:
     """异步示例任务，模拟耗时操作"""
-    await asleep(20)
+    await asleep(30)
     return 'test async'
+
+
+@celery_app.task(name='task_demo_params')
+async def task_demo_params(hello: str, world: str | None = None) -> str:
+    """参数示例任务，模拟传参操作"""
+    return hello + world

backend/app/task/utils/schedulers.py

@@ -258,8 +258,8 @@ async def _unpack_fields(
             except KeyError:
                 continue
         model_dict.update(
-            args=json.dumps(args) if args else None,
-            kwargs=json.dumps(kwargs) if kwargs else None,
+            args=json.dumps(args, ensure_ascii=False) if args else None,
+            kwargs=json.dumps(kwargs, ensure_ascii=False) if kwargs else None,
             **cls._unpack_options(**options or {}),
             **entry,
         )