任务调度器¶

from scheduler import Scheduler, every

sched = Scheduler()
sched.add_job(lambda: print("tick"), every(10, "seconds"), id="ticker")
sched.start()

from scheduler import Scheduler, cron

sched = Scheduler()
sched.add_job(cleanup, cron("0 3 * * *"))       # 每天 03:00
sched.add_job(report, cron("0 9 * * 1-5"))      # 工作日 09:00
sched.add_job(backup, cron("30 2 1 * *"))       # 每月 1 号 02:30
sched.start()

from datetime import datetime, timedelta
from scheduler import Scheduler, once

sched = Scheduler()
run_at = datetime.now() + timedelta(hours=1)
sched.add_job(send_reminder, once(run_at), id="reminder")
sched.start()

import time
from scheduler import Scheduler, every

with Scheduler() as sched:
    sched.add_job(heartbeat, every(1, "seconds"))
    time.sleep(10)  # 运行 10 秒后自动关闭

from scheduler import Scheduler, every

sched = Scheduler()

@sched.scheduled_job(every(30, "seconds"), id="health")
def health_check():
    print("healthy")

sched.start()

import asyncio
from scheduler import Scheduler, every

sched = Scheduler()

async def fetch_data():
    await asyncio.sleep(0.1)
    print("fetched")

sched.add_job(fetch_data, every(60, "seconds"))
sched.start()

from scheduler import Scheduler, every

def on_ok(result):
    print(f"成功：{result}")

def on_fail(exc):
    print(f"错误：{exc}")

sched = Scheduler()
sched.add_job(
    risky_task,
    every(30, "seconds"),
    on_success=on_ok,
    on_error=on_fail,
)
sched.start()

from scheduler import Scheduler, every, EventType

sched = Scheduler()

def on_event(event):
    if event.event_type == EventType.job_executed:
        print(f"任务 {event.job_id} 在 {event.run_time} 执行")
    elif event.event_type == EventType.job_error:
        print(f"任务 {event.job_id} 失败：{event.exception}")

sched.add_listener(on_event)
sched.add_job(my_task, every(10, "seconds"))
sched.start()

from scheduler import Scheduler, every

sched = Scheduler()
sched.add_job(task_a, every(60, "seconds"), id="a")
sched.add_job(task_b, every(120, "seconds"), id="b")

sched.pause_job("a")          # 暂停任务 "a"
sched.resume_job("a")         # 恢复任务 "a"
sched.run_job("a")            # 立即执行
sched.remove_job("b")         # 删除任务 "b"
print(sched.get_jobs())       # 列出所有任务

from scheduler import Scheduler, every

# 如果任务延迟超过 5 秒，跳过而不执行
sched = Scheduler()
sched.add_job(
    sensitive_task,
    every(10, "seconds"),
    misfire_grace_time=5.0,
)
sched.start()