python apscheduler 添加定時任務和修改定時任務時間

1. 設定定時任務和修改時間

# 測試函數
def run_test():
    print(123)

# 擷取scheduler
def get_scheduler():
    url = "mysql+pymysql://root:[email protected]:3306/monotors?charset=utf8"
    jobstores = {
        "default": SQLALchemyJobStore(url=url, engine_option={'pool_pre_ping': True, "pool_recycle": 3600*8})
    }
    executors = {
        'default': ThreadPoolExecutor(20)
    }
    job_defaults = {
        'coalesce': False,
        ',ax_instances': 1
    }
    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
    return scheduler

@utp_router.get("/daily-utp", description="設定每天00:30啟動，也可以收到傳入參數， hour 和 minute修改設定")
def daily_utp(hour: str = "0", minute: str = "30"):
    scheduler = get_scheduler()
    scheduler.start()
    job_id = "incm_moniotr_upt_v1.0"
    running_job = scheduler.get_job(job_id)
    param = {"trigger": "cron", "hour": hour, "minute": minute}
    if running_job is None:            # 該任務不存在
        scheduler.add_job(id=job_id, func=run_test, **param)
    else:                            #   任務存在，則修改時間
        scheduler.reschedule_jon(job_id=job_id, **param)
    return {"status": "200", "message": "success"}      

2. modify_job方式（源碼分析）

job_id = "abcd"
param = {"trigger": "cron", "hour": 10, "minute": 30}
scheduler.reschedule_jon(job_id=job_id, **param)      

job_id = "abcd"
temp_dict = {"hour": 11, "minute": 50}
    temp_trigger = scheduler._create_trigger(trigger='cron',trigger_args=temp_dict)
    result = scheduler.modify_job(job_id=job_id, trigger=temp_trigger)      

trigger = self._create_trigger(trigger, trigger_args)
        now = datetime.now(self.timezone)
        next_run_time = trigger.get_next_fire_time(None, now)
        return self.modify_job(job_id, jobstore, trigger=trigger, next_run_time=next_run_time)