mirror of
https://github.com/fastapi-practices/fastapi-best-architecture.git
synced 2026-09-21 13:12:24 +00:00
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from celery import schedules
|
|
from celery.schedules import ParseException
|
|
|
|
from backend.common.exception import errors
|
|
from backend.utils.timezone import timezone
|
|
|
|
|
|
class TzAwareCrontab(schedules.crontab):
|
|
"""时区感知 Crontab"""
|
|
|
|
def __init__(self, minute='*', hour='*', day_of_week='*', day_of_month='*', month_of_year='*', app=None) -> None: # noqa: ANN001
|
|
super().__init__(
|
|
minute=minute,
|
|
hour=hour,
|
|
day_of_week=day_of_week,
|
|
day_of_month=day_of_month,
|
|
month_of_year=month_of_year,
|
|
nowfun=timezone.now,
|
|
app=app,
|
|
)
|
|
|
|
|
|
def crontab_verify(crontab: str) -> None:
|
|
"""
|
|
验证标准 crontab 表达式
|
|
|
|
:param crontab: 标准 crontab 表达式
|
|
:return:
|
|
"""
|
|
crontab_split = crontab.split(' ')
|
|
if len(crontab_split) != 5:
|
|
raise errors.RequestError(msg='Crontab 表达式非法')
|
|
try:
|
|
TzAwareCrontab.from_string(crontab)
|
|
except (ParseException, ValueError):
|
|
raise errors.RequestError(msg='Crontab 表达式非法')
|