Method of realizing timing task using APScheduler3.0. 1

  • 2021-07-24 11:14:57
  • OfStack

Requirement is to perform an action at a time specified by 1

Most of the online suggestions are implemented by calling add_date_job of Scheduler

However, APScheduler 3.0. 1 is quite different from the previous one and cannot be realized by the above method

Refer to userguide at https://apscheduler.readthedocs.org/en/latest/userguide. html APScheduler 3.0. 1 for resolution


from datetime import datetime
import time
import os
 
from apscheduler.schedulers.background import BackgroundScheduler
 
 
def tick():
 print('Tick! The time is: %s' % datetime.now())
 
 
if __name__ == '__main__':
 scheduler = BackgroundScheduler()
 scheduler.add_job(tick, 'interval', seconds=3)
 scheduler.start()
 print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
 
 try:
  # This is here to simulate application activity (which keeps the main thread alive).
  while True:
   time.sleep(2)
 except (KeyboardInterrupt, SystemExit):
  scheduler.shutdown() # Not strictly necessary if daemonic mode is enabled but should be done if possible

The code implementation of the instance executes the tick method every 3 seconds. Although it does not meet the requirements, it is found that add_interval_job has been implemented in APScheduler 3.0. 1


scheduler.add_job(tick, 'interval', seconds=3)

Replace.

help (scheduler.add_job) is obtained


add_job(func, trigger=None, args=None, kwargs=None, id=None, name=None, misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined, next_run_time=undefined, jobstore='default', executor='default', replace_existing=False, **trigger_args)
Adds the given job to the job list and wakes up the scheduler if it's already running.
 
Any option that defaults to undefined will be replaced with the corresponding default value when the job is scheduled (which happens when the scheduler is started, or immediately if the scheduler is already running).
 
The func argument can be given either as a callable object or a textual reference in the package.module:some.object format, where the first half (separated by :) is an importable module and the second half is a reference to the callable object, relative to the module.
 
The trigger argument can either be:
the alias name of the trigger (e.g. date, interval or cron), in which case any extra keyword arguments to this method are passed on to the trigger's constructor
an instance of a trigger class

Therefore, the first parameter is trigger, and the desirable values are date, interval, cron, and **trigger_args are the constructors of trigger.

Find the constructor of DateTrigger through the source code


def __init__(self, run_date=None, timezone=None)

So, just pass the specified time into add_job


scheduler.add_job(tick, 'date', run_date='2014-11-11 14:48:00')

Related articles: