From 98ef07ad3240a46f954117a5793f6c66e17e65a7 Mon Sep 17 00:00:00 2001 From: Wu Clan Date: Tue, 15 Jul 2025 00:28:10 +0800 Subject: [PATCH] Simplify celery task crontab config (#722) --- backend/app/task/model/scheduler.py | 10 +--- backend/app/task/schema/scheduler.py | 10 ++-- backend/app/task/service/scheduler_service.py | 28 ++++++----- backend/app/task/utils/schedulers.py | 46 +++++++++---------- 4 files changed, 44 insertions(+), 50 deletions(-) diff --git a/backend/app/task/model/scheduler.py b/backend/app/task/model/scheduler.py index b9be7dbb..45488638 100644 --- a/backend/app/task/model/scheduler.py +++ b/backend/app/task/model/scheduler.py @@ -41,15 +41,7 @@ class TaskScheduler(Base): type: Mapped[int] = mapped_column(comment='调度类型（0间隔 1定时）') interval_every: Mapped[int | None] = mapped_column(comment='任务再次运行前的间隔周期数') interval_period: Mapped[str | None] = mapped_column(String(255), comment='任务运行之间的周期类型') - crontab_minute: Mapped[str | None] = mapped_column(String(60 * 4), default='*', comment='运行的分钟，"*" 表示全部') - crontab_hour: Mapped[str | None] = mapped_column(String(24 * 4), default='*', comment='运行的小时，"*" 表示全部') - crontab_day_of_week: Mapped[str | None] = mapped_column(String(64), default='*', comment='运行的星期，"*" 表示全部') - crontab_day_of_month: Mapped[str | None] = mapped_column( - String(31 * 4), default='*', comment='运行的每月日期，"*" 表示全部' - ) - crontab_month_of_year: Mapped[str | None] = mapped_column( - String(64), default='*', comment='运行的月份，"*" 表示全部' - ) + crontab: Mapped[str | None] = mapped_column(String(50), default='* * * * *', comment='任务运行的 Crontab 计划') one_off: Mapped[bool] = mapped_column( Boolean().with_variant(INTEGER, 'postgresql'), default=False, comment='是否仅运行一次' ) diff --git a/backend/app/task/schema/scheduler.py b/backend/app/task/schema/scheduler.py index 3a03d973..8101c8b9 100644 --- a/backend/app/task/schema/scheduler.py +++ b/backend/app/task/schema/scheduler.py @@ -14,8 +14,8 @@ class TaskSchedulerSchemeBase(SchemaBase): name: str = Field(description='任务名称') task: str = Field(description='要运行的 Celery 任务') - args: JsonValue | None = Field(default='[]', description='任务可接收的位置参数') - kwargs: JsonValue | None = Field(default='{}', description='任务可接收的关键字参数') + args: JsonValue | None = Field(default=None, description='任务可接收的位置参数') + kwargs: JsonValue | None = Field(default=None, 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 路由的路由密钥') @@ -25,11 +25,7 @@ class TaskSchedulerSchemeBase(SchemaBase): type: TaskSchedulerType = Field(default=TaskSchedulerType.INTERVAL, description='任务调度类型（0间隔 1定时）') interval_every: int | None = 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='运行的星期，"*" 表示全部') - crontab_day_of_month: str | None = Field(default='*', description='运行的每月日期，"*" 表示全部') - crontab_month_of_year: str | None = Field(default='*', description='运行的月份，"*" 表示全部') + crontab: str = Field(default='* * * * *', description='运行的 Crontab 表达式') one_off: bool = Field(default=False, description='是否仅运行一次') remark: str | None = Field(default=None, description='备注') diff --git a/backend/app/task/service/scheduler_service.py b/backend/app/task/service/scheduler_service.py index b48b4d2b..712fd8d1 100644 --- a/backend/app/task/service/scheduler_service.py +++ b/backend/app/task/service/scheduler_service.py @@ -64,6 +64,15 @@ class TaskSchedulerService: task_scheduler = await task_scheduler_dao.get_by_name(db, obj.name) if task_scheduler: raise errors.ConflictError(msg='任务调度已存在') + if obj.type == TaskSchedulerType.CRONTAB: + crontab_split = obj.crontab.split(' ') + if len(crontab_split) != 5: + raise errors.RequestError(msg='Crontab 表达式非法') + crontab_verify('m', crontab_split[0]) + crontab_verify('h', crontab_split[1]) + crontab_verify('dow', crontab_split[2]) + crontab_verify('dom', crontab_split[3]) + crontab_verify('moy', crontab_split[4]) await task_scheduler_dao.create(db, obj) @staticmethod @@ -83,11 +92,14 @@ class TaskSchedulerService: if await task_scheduler_dao.get_by_name(db, obj.name): raise errors.ConflictError(msg='任务调度已存在') if task_scheduler.type == TaskSchedulerType.CRONTAB: - crontab_verify('m', task_scheduler.crontab_minute) - crontab_verify('h', task_scheduler.crontab_hour) - crontab_verify('dow', task_scheduler.crontab_day_of_week) - crontab_verify('dom', task_scheduler.crontab_day_of_month) - crontab_verify('moy', task_scheduler.crontab_month_of_year) + crontab_split = obj.crontab.split(' ') + if len(crontab_split) != 5: + raise errors.RequestError(msg='Crontab 表达式非法') + crontab_verify('m', crontab_split[0]) + crontab_verify('h', crontab_split[1]) + crontab_verify('dow', crontab_split[2]) + crontab_verify('dom', crontab_split[3]) + crontab_verify('moy', crontab_split[4]) count = await task_scheduler_dao.update(db, pk, obj) return count @@ -103,12 +115,6 @@ class TaskSchedulerService: task_scheduler = await task_scheduler_dao.get(db, pk) if not task_scheduler: raise errors.NotFoundError(msg='任务调度不存在') - if task_scheduler.type == TaskSchedulerType.CRONTAB: - crontab_verify('m', task_scheduler.crontab_minute) - crontab_verify('h', task_scheduler.crontab_hour) - crontab_verify('dow', task_scheduler.crontab_day_of_week) - crontab_verify('dom', task_scheduler.crontab_day_of_month) - crontab_verify('moy', task_scheduler.crontab_month_of_year) count = await task_scheduler_dao.set_status(db, pk, not task_scheduler.enabled) return count diff --git a/backend/app/task/utils/schedulers.py b/backend/app/task/utils/schedulers.py index 7005631f..b6a1119f 100644 --- a/backend/app/task/utils/schedulers.py +++ b/backend/app/task/utils/schedulers.py @@ -47,13 +47,14 @@ class ModelEntry(ScheduleEntry): and model.interval_period is not None ): self.schedule = schedules.schedule(timedelta(**{model.interval_period: model.interval_every})) - elif model.type == TaskSchedulerType.CRONTAB and model.crontab_minute is not None: + elif model.type == TaskSchedulerType.CRONTAB and model.crontab is not None: + crontab_split = model.crontab.split(' ') self.schedule = TzAwareCrontab( - minute=model.crontab_minute, - hour=model.crontab_hour or '*', - day_of_week=model.crontab_day_of_week or '*', - day_of_month=model.crontab_day_of_month or '*', - month_of_year=model.crontab_month_of_year or '*', + minute=crontab_split[0], + hour=crontab_split[1], + day_of_week=crontab_split[2], + day_of_month=crontab_split[3], + month_of_year=crontab_split[4], ) else: raise errors.NotFoundError(msg=f'{self.name} 计划为空！') @@ -63,8 +64,8 @@ class ModelEntry(ScheduleEntry): asyncio.create_task(self._disable(model)) try: - self.args = json.loads(model.args) if model.args else [] - self.kwargs = json.loads(model.kwargs) if model.kwargs else {} + self.args = json.loads(model.args) if model.args else None + self.kwargs = json.loads(model.kwargs) if model.kwargs else None except ValueError as exc: logger.error(f'禁用参数错误的任务：{self.name}；error: {str(exc)}') asyncio.create_task(self._disable(model)) @@ -187,22 +188,21 @@ class ModelEntry(ScheduleEntry): if not obj: obj = TaskScheduler(**CreateTaskSchedulerParam(task=task, **spec).model_dump()) elif isinstance(schedule, schedules.crontab): + crontab_minute = schedule._orig_minute if crontab_verify('m', schedule._orig_minute, False) else '*' + crontab_hour = schedule._orig_hour if crontab_verify('h', schedule._orig_hour, False) else '*' + crontab_day_of_week = ( + schedule._orig_day_of_week if crontab_verify('dom', schedule._orig_day_of_week, False) else '*' + ) + crontab_day_of_month = ( + schedule._orig_day_of_month if crontab_verify('dom', schedule._orig_day_of_month, False) else '*' + ) + crontab_month_of_year = ( + schedule._orig_month_of_year if crontab_verify('moy', schedule._orig_month_of_year, False) else '*' + ) spec = { 'name': name, 'type': TaskSchedulerType.CRONTAB.value, - 'crontab_minute': schedule._orig_minute - if crontab_verify('m', schedule._orig_minute, False) - else '*', - 'crontab_hour': schedule._orig_hour if crontab_verify('h', schedule._orig_hour, False) else '*', - 'crontab_day_of_week': schedule._orig_day_of_week - if crontab_verify('dom', schedule._orig_day_of_week, False) - else '*', - 'crontab_day_of_month': schedule._orig_day_of_month - if crontab_verify('dom', schedule._orig_day_of_month, False) - else '*', - 'crontab_month_of_year': schedule._orig_month_of_year - if crontab_verify('moy', schedule._orig_month_of_year, False) - else '*', + 'crontab': f'{crontab_minute} {crontab_hour} {crontab_day_of_week} {crontab_day_of_month} {crontab_month_of_year}', # noqa: E501 } stmt = select(TaskScheduler).filter_by(**spec) query = await db.execute(stmt) @@ -233,8 +233,8 @@ class ModelEntry(ScheduleEntry): except KeyError: continue model_dict.update( - args=json.dumps(args or []), - kwargs=json.dumps(kwargs or {}), + args=json.dumps(args) if args else None, + kwargs=json.dumps(kwargs) if kwargs else None, **cls._unpack_options(**options or {}), **entry, )