Decorators API¶
装饰器 API 提供了简洁优雅的任务定义方式。
基础装饰器¶
scheduled¶
scheduled(*, name=None, cron=None, interval=None, start_time=None, end_time=None, max_instances=1, timeout=None, retry_policy=None, enabled=True, tags=None, metadata=None)
¶
通用定时任务装饰器。
参数
name: 任务名称(默认使用函数名) cron: Cron 表达式 interval: 间隔时间(秒或 timedelta) start_time: 开始时间 end_time: 结束时间 max_instances: 最大并发实例数 timeout: 超时时间(秒) retry_policy: 重试策略 enabled: 是否启用 tags: 标签列表 metadata: 元数据
返回值
装饰器函数
示例
@scheduled(cron="/5 * * * * ") # 每5秒执行 async def my_task(): print("执行任务")
@scheduled(interval=60, retry_policy=RetryPolicy.aggressive()) async def another_task(): print("每60秒执行一次")
cron¶
cron(expression, *, name=None, retry_policy=None, timeout=None, **kwargs)
¶
Cron 表达式定时任务装饰器。
参数
expression: Cron 表达式(支持秒级) name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@cron("/10 * * * * ") # 每10秒执行 async def sync_data(): await fetch_and_sync()
@cron("0 0 * * *") # 每天零点执行 async def daily_cleanup(): await cleanup_old_data()
interval¶
interval(seconds, *, name=None, retry_policy=None, timeout=None, **kwargs)
¶
固定间隔定时任务装饰器。
参数
seconds: 间隔秒数或 timedelta name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@interval(30) # 每30秒执行 async def check_health(): await ping_services()
@interval(timedelta(hours=1)) # 每小时执行 async def hourly_report(): await generate_report()
once¶
once(at, *, name=None, retry_policy=None, timeout=None, **kwargs)
¶
一次性定时任务装饰器。
参数
at: 执行时间 name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@once(datetime(2024, 12, 31, 23, 59, 59)) async def new_year_task(): print("新年快乐!")
便捷装饰器¶
every¶
every(seconds=None, minutes=None, hours=None, days=None, *, name=None, retry_policy=None, timeout=None, **kwargs)
¶
更直观的间隔任务装饰器。
参数
seconds: 秒数 minutes: 分钟数 hours: 小时数 days: 天数 name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@every(minutes=30) # 每 30 分钟 async def half_hourly_task(): print("每 30 分钟执行")
@every(hours=2, minutes=30) # 每 2.5 小时 async def custom_interval(): print("每 2.5 小时执行")
hourly¶
hourly(*, minute=0, name=None, retry_policy=None, timeout=None, **kwargs)
¶
每小时执行的任务装饰器。
参数
minute: 在第几分钟执行 (0-59),默认为整点 name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@hourly() # 每小时整点执行 async def hourly_task(): print("每小时执行")
@hourly(minute=30) # 每小时的第 30 分钟执行 async def half_past(): print("每小时 30 分执行")
daily¶
daily(*, hour=0, minute=0, name=None, retry_policy=None, timeout=None, **kwargs)
¶
每天执行的任务装饰器。
参数
hour: 小时 (0-23),默认为 0 点 minute: 分钟 (0-59),默认为 0 分 name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@daily(hour=9, minute=30) # 每天 9:30 async def morning_task(): print("早上 9:30 执行")
@daily() # 每天 0:00 async def midnight_task(): print("每天午夜执行")
weekly¶
weekly(*, day=0, hour=0, minute=0, name=None, retry_policy=None, timeout=None, **kwargs)
¶
每周执行的任务装饰器。
参数
day: 星期几 (0=周日, 1=周一, ..., 6=周六),默认为周日 hour: 小时 (0-23) minute: 分钟 (0-59) name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@weekly(day=1, hour=9) # 每周一 9:00 async def weekly_report(): print("周一报表")
@weekly(day=5, hour=18) # 每周五 18:00 async def tgif(): print("周五晚上!")
monthly¶
monthly(*, day=1, hour=0, minute=0, name=None, retry_policy=None, timeout=None, **kwargs)
¶
每月执行的任务装饰器。
参数
day: 几号 (1-31),默认为 1 号 hour: 小时 (0-23) minute: 分钟 (0-59) name: 任务名称 retry_policy: 重试策略 timeout: 超时时间 **kwargs: 其他参数
返回值
装饰器函数
示例
@monthly(day=1, hour=0) # 每月 1 号 0:00 async def monthly_cleanup(): print("月初清理")
@monthly(day=15) # 每月 15 号 0:00 async def mid_month(): print("月中任务")