4

Using APScheduler version 3.0.3. Services in my application internally use APScheduler to schedule & run jobs. Also I did create a wrapper class around the actual APScheduler(just a façade, helps in unit tests). For unit testing these services, I can mock the this wrapper class. But I have a situation where I would really like the APScheduler to run the job (during test). Is there any way by which one can force run the job?

amrx
  • 673
  • 1
  • 9
  • 23

3 Answers3

4

There is no default trigger to launch a job immediately, to achieve this you can get the current time and set a DateTrigger to the job like this:

my_job.modify_job(trigger=DateTrigger(run_date=datetime.datetime.now()))

this way you will "force" the job to run, but you have to make sure to insert the job again in the scheduler, another option is just creating a new job to run the same function with the add_job function

sched.add_job(func=your_function(),
              trigger=DateTrigger(run_date=datetime.datetime.now()))

this way you don't have to do any additional step.

Jesus Gomez
  • 1,460
  • 13
  • 30
0

Another approach: you can write logic of your job in separate function. So, you will be able to call this function in your scheduled job as well as somewhere else. I guess that this is a more explicit way to do what you want.

Mark Mishyn
  • 3,921
  • 2
  • 28
  • 30
  • This has the disadvantage of allowing a concurrent run of your code in case the scheduler happens to "wake up" when you are calling your function manually. – tggm May 31 '19 at 11:01
0

Two alternatives.

  1. change the jobs next_run_time(not verified personally, but here says it works: APScheduler how to trigger job now)

    job = self.scheduler.get_job(job_id=job_id) job.modify(next_run_time=datetime.now() + timedelta(seconds=5))

  2. Add a run-once job to trigger it(verified):

    self.scheduler.add_job(func=job.run, executor=executor, trigger="date", max_instances=1, run_date=datetime.now() + timedelta(seconds=5))

Levin
  • 298
  • 3
  • 8