public/fastapi-best-architecture
1
0
Fork 0
mirror of https://github.com/fastapi-practices/fastapi-best-architecture.git synced 2026-09-21 13:12:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify celery task crontab config (#722)

Browse Source
This commit is contained in:
Wu Clan
2025-07-15 00:28:10 +08:00
committed by GitHub
parent 802b0d456a
commit 98ef07ad32
4 changed files with 44 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/task/model/scheduler.py
+1 -9
View File
@@ -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='是否仅运行一次'
)
backend/app/task/schema/scheduler.py
+3 -7
View File
@@ -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='备注')
backend/app/task/service/scheduler_service.py
+17 -11
View File
@@ -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
backend/app/task/utils/schedulers.py
+23 -23
View File
@@ -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,
)