如何正确获取APScheduler任务的next_run_time?遇AttributeError
问题分析与解决
错误原因
你在调度器启动前就尝试访问job.next_run_time属性。APScheduler中,任务的next_run_time是由调度器的调度循环计算生成的,在sched.start()调用前,调度器未启动,任务还未被调度处理,该属性尚未初始化,因此触发AttributeError。
正确获取方式
方式一:启动前手动计算下次运行时间
调用任务的compute_next_run_time()方法,手动触发计算逻辑,即可在调度器启动前获取next_run_time:
sched.add_job(myFunc, 'cron', minute='*', start_date=datetime(2023,1,1), end_date=datetime(2023,1,3)) for job in sched.get_jobs(): job.compute_next_run_time() # 手动计算下次运行时间 print(job.next_run_time) sched.start()
方式二:启动后获取next_run_time
由于sched.start()是阻塞调用,需要通过线程等异步方式在调度器启动后执行获取逻辑:
import threading import time def fetch_next_run_times(sched): time.sleep(0.5) # 等待调度器完成初始化 for job in sched.get_jobs(): print(job.next_run_time) sched.add_job(myFunc, 'cron', minute='*', start_date=datetime(2023,1,1), end_date=datetime(2023,1,3)) # 启动异步线程执行获取操作 threading.Thread(target=fetch_next_run_times, args=(sched,)).start() sched.start()
内容的提问来源于stack exchange,提问作者Bijan
相关产品推荐
相关产品推荐

